Python os.readv 函数

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

这份全面的指南探索了 Python 的 os.readv 函数，该函数执行从文件描述符到多个缓冲区的分散读取。我们将介绍它的 Unix 起源、性能优势和实际示例。

基本定义

os.readv 函数通过单个系统调用将数据从文件描述符读取到多个缓冲区中。 这被称为分散或向量化 I/O。

关键参数：fd (文件描述符), buffers (可写缓冲区的序列)。 返回读取的总字节数。 需要具有 readv() 系统调用的类 Unix 系统。

基本 readv 示例

此示例演示了 os.readv 最简单的用法，即从文件读取数据到两个单独的缓冲区中。 我们首先写入一些测试数据。

import os

# Create test file
with open("test.dat", "wb") as f:
    f.write(b"HelloWorldPythonReadv")

# Open file and prepare buffers
fd = os.open("test.dat", os.O_RDONLY)
buf1 = bytearray(5)  # First 5 bytes
buf2 = bytearray(10) # Next 10 bytes

# Perform scatter read
buffers = [buf1, buf2]
bytes_read = os.readv(fd, buffers)

print(f"Total bytes read: {bytes_read}")
print(f"Buffer 1: {buf1.decode()}")
print(f"Buffer 2: {buf2.decode()}")

os.close(fd)

此示例将数据写入文件，然后将其读回两个单独的缓冲区。 第一个缓冲区获取 5 个字节，第二个缓冲区获取 10 个字节。

os.readv 调用在一个操作中填充两个缓冲区，这比多个读取调用更有效。

读取到多个缓冲区

os.readv 可以读取到任意数量的缓冲区中。 此示例显示了从二进制文件读取到三个大小不同的缓冲区中。

import os

# Generate binary data (24 bytes)
data = bytes(range(24))
with open("data.bin", "wb") as f:
    f.write(data)

fd = os.open("data.bin", os.O_RDONLY)

# Three buffers of different sizes
buf1 = bytearray(8)  # First 8 bytes
buf2 = bytearray(12) # Next 12 bytes
buf3 = bytearray(4)  # Final 4 bytes

buffers = [buf1, buf2, buf3]
bytes_read = os.readv(fd, buffers)

print(f"Read {bytes_read} bytes total")
print(f"Buffer 1: {list(buf1)}")
print(f"Buffer 2: {list(buf2)}")
print(f"Buffer 3: {list(buf3)}")

os.close(fd)

这将创建一个包含 24 个字节 (0-23) 的二进制文件，然后将其读取到三个大小不同的缓冲区中。 输出显示每个缓冲区中的字节值。

读取的总字节数将是缓冲区大小的总和，除非到达 EOF。

使用 readv 进行部分读取

当文件小于总缓冲区容量时，os.readv 执行部分读取。 此示例演示了如何处理此类情况。

import os

# Create small file (10 bytes)
with open("small.dat", "wb") as f:
    f.write(b"ShortFile")

fd = os.open("small.dat", os.O_RDONLY)

# Buffers that total more than file size
buf1 = bytearray(6)
buf2 = bytearray(6)
buf3 = bytearray(6)

buffers = [buf1, buf2, buf3]
bytes_read = os.readv(fd, buffers)

print(f"Read {bytes_read} bytes (file has 10 bytes)")
print(f"Buffer 1: {buf1.decode()}")
print(f"Buffer 2: {buf2.decode()}")
print(f"Buffer 3: {buf3.decode()}")  # Will be empty

os.close(fd)

此示例显示了当从一个小文件读取到更大的缓冲区中时会发生什么。 只有前几个缓冲区被填充，并且读取的总字节数与文件大小匹配。

第三个缓冲区保持为空，因为文件内容适合前两个缓冲区。

从不可查找的文件读取

os.readv 可以与管道和套接字等不可查找的文件一起使用。 此示例从管道读取到多个缓冲区中。

import os

# Create pipe
r, w = os.pipe()

# Write data to pipe
os.write(w, b"PipeDataForReadvExample")

# Prepare buffers
buf1 = bytearray(8)
buf2 = bytearray(12)

# Read from pipe
buffers = [buf1, buf2]
bytes_read = os.readv(r, buffers)

print(f"Read {bytes_read} bytes from pipe")
print(f"Buffer 1: {buf1.decode()}")
print(f"Buffer 2: {buf2.decode()}")

os.close(r)
os.close(w)

这将创建一个管道，将数据写入其中，然后使用 os.readv 读取。 相同的方法适用于套接字和其他不可查找的文件描述符。

对于网络编程，readv 可以有效地将传入的数据收集到预分配的缓冲区中。

性能比较

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

import os
import time

# Create large test file (1MB)
data = os.urandom(1024 * 1024)
with open("large.dat", "wb") as f:
    f.write(data)

def test_readv():
    fd = os.open("large.dat", os.O_RDONLY)
    buf1 = bytearray(512 * 1024)
    buf2 = bytearray(512 * 1024)
    os.readv(fd, [buf1, buf2])
    os.close(fd)

def test_multiple_read():
    fd = os.open("large.dat", os.O_RDONLY)
    buf1 = bytearray(512 * 1024)
    buf2 = bytearray(512 * 1024)
    os.read(fd, buf1)
    os.read(fd, buf2)
    os.close(fd)

# Time readv
start = time.time()
for _ in range(100):
    test_readv()
readv_time = time.time() - start

# Time multiple reads
start = time.time()
for _ in range(100):
    test_multiple_read()
read_time = time.time() - start

print(f"readv time: {readv_time:.3f}s")
print(f"Multiple read time: {read_time:.3f}s")
print(f"Ratio: {read_time/readv_time:.1f}x")

此基准测试创建一个 1MB 的文件，并使用两种方法读取 100 次。 由于系统调用较少，os.readv 通常表现出更好的性能。

确切的加速取决于系统和文件大小，但对于分散的读取，向量化 I/O 通常更有效。

使用 readv 进行错误处理

此示例演示了使用 os.readv 时的正确错误处理，包括处理部分读取和系统错误。

import os
import errno

try:
    # Try reading from invalid FD
    buf = bytearray(10)
    os.readv(999, [buf])
except OSError as e:
    if e.errno == errno.EBADF:
        print("Error: Bad file descriptor")
    else:
        print(f"OS error: {e}")

# Handle partial read
try:
    fd = os.open("empty.dat", os.O_CREAT | os.O_RDONLY)
    buf1 = bytearray(10)
    buf2 = bytearray(10)
    bytes_read = os.readv(fd, [buf1, buf2])
    print(f"Read {bytes_read} bytes from empty file")
    os.close(fd)
except OSError as e:
    print(f"Error: {e}")
finally:
    if 'fd' in locals():
        os.close(fd)

第一部分尝试从无效的文件描述符读取，显示了如何处理特定错误。 第二部分演示了从空文件读取。

在处理底层文件操作时，正确的错误处理至关重要，因为可能会出现很多问题。

高级：使用 readv 的 Memoryview

为了获得最大效率，os.readv 可以与 memoryview 对象一起使用，以避免不必要的复制。 此示例显示了优化的方法。

import os

# Create test file
data = b"MemoryViewExampleData"
with open("memview.dat", "wb") as f:
    f.write(data)

fd = os.open("memview.dat", os.O_RDONLY)

# Allocate memory and create memoryviews
buffer = bytearray(20)
mv1 = memoryview(buffer)[0:10]  # First 10 bytes
mv2 = memoryview(buffer)[10:20] # Next 10 bytes

# Read into memoryviews
bytes_read = os.readv(fd, [mv1, mv2])

print(f"Read {bytes_read} bytes")
print(f"Full buffer: {buffer.decode()}")
print(f"View 1: {mv1.tobytes().decode()}")
print(f"View 2: {mv2.tobytes().decode()}")

os.close(fd)

此示例分配一个缓冲区，然后创建两个引用其部分的 memoryview 对象。 os.readv 直接写入这些视图。

使用 memoryview 可以避免额外的内存分配和复制，这对于高性能应用程序非常重要。

性能注意事项

• 更少的系统调用： readv 将多个读取合并为一个
• 缓冲区管理： 预分配减少开销
• 内存效率： Memoryview 阻止复制
• 对齐： 正确的缓冲区对齐提高了性能
• 内核绕过： 某些系统优化了向量化 I/O

最佳实践

• 缓冲区大小调整： 根据预期的数据调整缓冲区大小
• 错误处理： 始终检查返回值和错误
• 资源清理： 使用 try/finally 处理文件描述符
• 可移植性： 检查系统支持 (主要是 Unix)
• 基准测试： 测试您的用例的性能提升

资料来源

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

作者

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

列出所有 Python 教程。