Python os.writev 函数

上次修改时间：2025 年 4 月 11 日

本篇综合指南探讨了 Python 的 os.writev 函数，该函数执行分散-聚集 I/O 操作。我们将介绍缓冲区管理、性能优势和实际的文件写入示例。

基本定义

os.writev 函数将来自多个缓冲区的数据写入单个系统调用中的文件描述符。这被称为矢量化 I/O 或分散-聚集 I/O。

关键参数：fd (文件描述符)，buffers (类字节对象的序列)。返回写入的总字节数。在类 Unix 系统（非 Windows）上可用。

使用 writev 的基本文件写入

此示例演示了使用 os.writev 将多个缓冲区写入文件的最简单用法。我们将创建一个文件并写入三个单独的块。

import os

# Create a test file
fd = os.open("output.txt", os.O_WRONLY | os.O_CREAT, 0o644)

# Prepare multiple buffers
buffers = [
    b"First line\n",
    b"Second line\n",
    b"Third line\n"
]

# Write all buffers at once
bytes_written = os.writev(fd, buffers)
print(f"Total bytes written: {bytes_written}")

os.close(fd)

此代码打开一个文件，创建三个字节缓冲区，并在一个操作中写入它们。该函数返回所有缓冲区写入的总字节数。

与多个写入调用相比，其优势在于减少了系统调用开销，这对于高性能应用程序尤其有利。

写入不同的缓冲区类型

os.writev 接受各种类字节对象。此示例显示了写入不同的类型：bytes、bytearray 和 memoryview 对象。

import os

fd = os.open("mixed_buffers.bin", os.O_WRONLY | os.O_CREAT, 0o644)

buffers = [
    b"Static bytes\n",
    bytearray(b"Bytearray content\n"),
    memoryview(b"Memoryview content\n")
]

bytes_written = os.writev(fd, buffers)
print(f"Total bytes written: {bytes_written}")

os.close(fd)

这演示了 os.writev 对不同类字节对象的灵活性。所有这些都在单个系统调用中高效写入。

Memoryview 特别有用，因为它避免了复制大型缓冲区，同时仍可与矢量化写入操作一起使用。

性能比较

此示例将 os.writev 与多个 os.write 调用进行比较，以演示矢量化 I/O 的性能优势。

import os
import time

def test_writev(fd, buffers):
    start = time.perf_counter()
    os.writev(fd, buffers)
    return time.perf_counter() - start

def test_multiple_writes(fd, buffers):
    start = time.perf_counter()
    for buf in buffers:
        os.write(fd, buf)
    return time.perf_counter() - start

# Prepare test data
buffers = [b"x" * 1024 for _ in range(1000)]
fd = os.open("perf_test.bin", os.O_WRONLY | os.O_CREAT)

# Test writev
writev_time = test_writev(fd, buffers)
os.ftruncate(fd, 0)  # Reset file

# Test multiple writes
write_time = test_multiple_writes(fd, buffers)

os.close(fd)

print(f"writev time: {writev_time:.6f}s")
print(f"Multiple writes time: {write_time:.6f}s")
print(f"Speedup: {write_time/writev_time:.1f}x")

此基准测试创建 1000 个缓冲区，并使用两种方法写入它们。在大多数系统上，os.writev 将显着更快。

性能差异来自减少的系统调用开销 - 一个调用而不是多个调用。随着缓冲区数量的增加，优势也随之增加。

使用 writev 的部分写入

与常规写入一样，os.writev 可能会执行部分写入。此示例演示了如何通过重试剩余数据来处理此类情况。

import os

def robust_writev(fd, buffers):
    total = 0
    while buffers:
        try:
            written = os.writev(fd, buffers)
            total += written
            # Adjust buffers based on what was written
            while written > 0:
                if written >= len(buffers[0]):
                    written -= len(buffers[0])
                    buffers.pop(0)
                else:
                    buffers[0] = buffers[0][written:]
                    written = 0
        except BlockingIOError:
            # Would block - wait and retry
            time.sleep(0.1)
        except InterruptedError:
            # System call interrupted - retry
            continue
    return total

fd = os.open("partial_test.txt", os.O_WRONLY | os.O_CREAT | os.O_NONBLOCK)
buffers = [b"a" * 10000, b"b" * 20000, b"c" * 30000]

bytes_written = robust_writev(fd, buffers)
print(f"Successfully wrote {bytes_written} bytes")

os.close(fd)

此实现通过跟踪写入了多少并相应地调整缓冲区来处理部分写入。它还处理非阻塞方案。

该函数会持续写入，直到所有缓冲区都耗尽，使其能够可靠地应对部分写入和中断。

网络套接字写入

os.writev 特别适用于网络编程。此示例演示了如何有效地将 HTTP 响应标头和正文写入套接字。

import os
import socket

def send_response(client_sock, headers, body):
    # Convert headers to bytes
    header_bytes = "\r\n".join(headers).encode("utf-8")
    
    # Prepare buffers - headers + blank line + body
    buffers = [
        header_bytes,
        b"\r\n\r\n",
        body if isinstance(body, (bytes, bytearray, memoryview)) else body.encode("utf-8")
    ]
    
    # Get raw file descriptor from socket
    fd = client_sock.fileno()
    
    # Write all at once
    try:
        total = os.writev(fd, buffers)
        print(f"Sent {total} bytes")
    except (BrokenPipeError, ConnectionResetError) as e:
        print(f"Connection error: {e}")

# Example usage
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("localhost", 8080))
server_socket.listen(1)

client, addr = server_socket.accept()
print(f"Connection from {addr}")

headers = [
    "HTTP/1.1 200 OK",
    "Content-Type: text/plain",
    "Connection: close"
]
body = "Hello from writev example!"

send_response(client, headers, body)
client.close()
server_socket.close()

这演示了 os.writev 如何通过将标头和正文组合在一个系统调用中来有效地发送 HTTP 响应。

与多个发送调用相比，该技术减少了延迟和 CPU 开销，这对于高性能服务器尤其重要。

Memoryview 的内存效率

此示例显示了如何将 memoryview 与 os.writev 一起使用，以在不复制数据的情况下写入大型文件。

import os

def write_large_file(filename, chunks):
    fd = os.open(filename, os.O_WRONLY | os.O_CREAT, 0o644)
    
    # Convert chunks to memoryviews to avoid copying
    buffers = [memoryview(chunk) for chunk in chunks]
    
    total = 0
    while buffers:
        written = os.writev(fd, buffers)
        total += written
        
        # Adjust buffers based on what was written
        while written > 0 and buffers:
            chunk_len = len(buffers[0])
            if written >= chunk_len:
                written -= chunk_len
                buffers.pop(0)
            else:
                buffers[0] = buffers[0][written:]
                written = 0
    
    os.close(fd)
    return total

# Generate 1MB of data in 10 chunks
data_chunks = [os.urandom(100000) for _ in range(10)]
bytes_written = write_large_file("large_file.bin", data_chunks)
print(f"Wrote {bytes_written} bytes")

此方法具有内存效率，因为 memoryview 允许切片而不复制底层数据。它非常适合大型文件。

该函数处理部分写入并跟踪缓冲区列表中的进度，确保最终写入所有数据。

错误处理

此示例演示了 os.writev 的全面错误处理，涵盖了常见的故障情况和边缘情况。

import os
import errno

def safe_writev(fd, buffers):
    if not buffers:
        raise ValueError("Empty buffer list")
    
    for buf in buffers:
        if not isinstance(buf, (bytes, bytearray, memoryview)):
            raise TypeError(f"Buffer must be bytes-like, got {type(buf)}")
    
    try:
        return os.writev(fd, buffers)
    except OSError as e:
        if e.errno == errno.EBADF:
            raise ValueError("Invalid file descriptor") from e
        elif e.errno == errno.EINVAL:
            raise ValueError("Invalid argument") from e
        elif e.errno == errno.EFAULT:
            raise ValueError("Bad buffer address") from e
        elif e.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
            raise BlockingIOError("Write would block") from e
        elif e.errno == errno.EINTR:
            raise InterruptedError("System call interrupted") from e
        raise  # Re-raise other errors

# Example usage
try:
    fd = os.open("error_test.txt", os.O_WRONLY | os.O_CREAT)
    buffers = [b"test", b"data"]
    
    try:
        written = safe_writev(fd, buffers)
        print(f"Wrote {written} bytes")
    finally:
        os.close(fd)
except Exception as e:
    print(f"Error: {e}")

此实现为 os.writev 可能遇到的各种故障情况提供了详细的错误检查和有意义的错误消息。

该函数验证输入缓冲区，并将系统错误转换为具有明确说明的适当的 Python 异常。

安全注意事项

• 文件描述符： 确保描述符有效且已正确打开
• 缓冲区验证： 验证所有缓冲区都是类字节对象
• 部分写入： 始终处理未写入所有数据的情况
• 资源限制： 大量写入可能会达到系统限制
• 平台支持： 在 Windows 系统上不可用

最佳实践

• 使用 memoryview： 对于大型缓冲区，避免复制
• 处理部分写入： 为提高鲁棒性，实施重试逻辑
• 基准测试： 针对您的用例与替代方案进行比较
• 错误处理： 涵盖所有可能的故障情况
• 记录假设： 清楚地说明平台要求

资料来源

• Python os.writev 文档
• Linux writev(2) 手册页

作者

我的名字是 Jan Bodnar，我是一位充满热情的程序员，拥有丰富的编程经验。 自 2007 年以来，我一直在撰写编程文章。 迄今为止，我撰写了超过 1,400 篇文章和 8 本电子书。 我拥有超过十年的编程教学经验。

列出所有 Python 教程。