Python os.pwrite 函数

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

这个综合指南探索了 Python 的 os.pwrite 函数，该函数在特定偏移量处将数据写入文件，而无需更改文件指针。我们将介绍它的参数、行为和实际用例。

基本定义

os.pwrite 函数在指定的偏移量处将字节写入文件描述符。它类似于 os.write，但不修改文件指针位置。

关键参数：fd (文件描述符)，data (要写入的字节)，offset (文件中的位置)。返回写入的字节数。需要打开文件以进行写入。

基本文件写入

此示例演示了 os.pwrite 的基本用法，用于在文件中的特定位置写入数据。我们首先创建一个包含一些内容的文件。

basic_write.py

import os

# Create a file with initial content
with open("data.txt", "w") as f:
    f.write("Initial content")

# Open file for writing and get file descriptor
fd = os.open("data.txt", os.O_RDWR)

# Write at offset 8 without changing file pointer
bytes_written = os.pwrite(fd, b"NEW ", 8)
print(f"Wrote {bytes_written} bytes")

# Verify content
os.lseek(fd, 0, os.SEEK_SET)
print(os.read(fd, 100))  # Output: b'Initial NEWtent'

os.close(fd)

这将在文件中位置 8 写入 "NEW "。原始内容 "content" 变为 "NEWtent"，因为我们覆盖了其中的一部分。在 pwrite 之后，文件指针保持不变。

请注意，我们使用 os.open 获取文件描述符，因为 pwrite 需要文件描述符，而不是文件对象。

在不同的偏移量处写入

此示例展示了在文件中多个位置写入，同时保持写入之间原始文件指针位置。

multiple_writes.py

import os

# Create a sample file
with open("records.txt", "w") as f:
    f.write("ABCDEFGHIJKLMNOPQRSTUVWXYZ")

fd = os.open("records.txt", os.O_RDWR)

# Get initial position
initial_pos = os.lseek(fd, 0, os.SEEK_CUR)
print(f"Initial position: {initial_pos}")

# Write at various offsets
os.pwrite(fd, b"123", 5)    # Overwrite positions 5-7
os.pwrite(fd, b"456", 10)   # Overwrite positions 10-12
os.pwrite(fd, b"789", 20)   # Overwrite positions 20-22

# Verify position hasn't changed
current_pos = os.lseek(fd, 0, os.SEEK_CUR)
print(f"Position after writes: {current_pos}")

# Read entire file
os.lseek(fd, 0, os.SEEK_SET)
print(os.read(fd, 100))  # Output: b'ABCDE123HIJ456NOPQRST789VWXYZ'

os.close(fd)

每个 pwrite 操作都在指定的偏移量处写入，而不影响其他操作或文件指针。这允许在已知文件位置进行精确修改。

输出显示原始字符串，并在指定位置插入了我们的数字序列。

使用 pwrite 追加

虽然 pwrite 通常用于覆盖，但我们可以通过在文件末尾写入来使用它进行追加。这需要知道文件大小。

append_write.py

import os

# Create initial file
with open("log.txt", "w") as f:
    f.write("Log starts here\n")

fd = os.open("log.txt", os.O_RDWR)

# Get file size for append
file_size = os.lseek(fd, 0, os.SEEK_END)

# Append new entries without moving pointer
os.pwrite(fd, b"Entry 1\n", file_size)
file_size += len("Entry 1\n")

os.pwrite(fd, b"Entry 2\n", file_size)
file_size += len("Entry 2\n")

os.pwrite(fd, b"Entry 3\n", file_size)

# Verify content
os.lseek(fd, 0, os.SEEK_SET)
print(os.read(fd, 1000).decode())

os.close(fd)

这演示了如何通过跟踪文件大小使用 pwrite 追加数据。每次写入都发生在当前内容的末尾。

虽然使用 O_APPEND 的 os.write 对于纯追加来说更简单，但这显示了 pwrite 的灵活性。

处理部分写入

pwrite 可能会写入少于请求的字节数。此示例展示了如何处理部分写入并确保写入完整数据。

partial_writes.py

import os

fd = os.open("partial.dat", os.O_RDWR | os.O_CREAT)

large_data = b"X" * 1000000  # 1MB of data
offset = 0
remaining = len(large_data)

while remaining > 0:
    written = os.pwrite(fd, large_data[-remaining:], offset)
    if written == 0:  # Disk full?
        raise IOError("Failed to write data")
    offset += written
    remaining -= written
    print(f"Wrote {written} bytes, {remaining} remaining")

os.close(fd)

# Verify
print(f"Final file size: {os.path.getsize('partial.dat')} bytes")

这实现了一个写入循环，该循环会一直持续到写入所有数据。它会更新每次部分写入的偏移量和剩余数据。

实际上，磁盘已满的情况很少见，但健壮的代码应处理大数据传输的部分写入。

线程安全的文件写入

当写入在单个文件系统块内时，pwrite 是原子的，这使其可用于线程安全的操作。此示例演示了此属性。

thread_safe.py

import os
import threading

fd = os.open("counter.txt", os.O_RDWR | os.O_CREAT)
os.write(fd, b"0")  # Initialize counter
os.lseek(fd, 0, os.SEEK_SET)

def increment_counter(thread_id):
    for _ in range(1000):
        # Read current value
        current = int(os.pread(fd, 1, 0))
        # Write incremented value
        os.pwrite(fd, str(current + 1).encode(), 0)

threads = []
for i in range(10):
    t = threading.Thread(target=increment_counter, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

os.lseek(fd, 0, os.SEEK_SET)
print(f"Final counter value: {os.read(fd, 10).decode()}")

os.close(fd)

这实现了一个简单的计数器，带有并发增量。虽然不完美（读取和写入之间存在竞争条件），但它演示了 pwrite 的原子写入能力。

对于真正的原子增量，除了 pwrite 之外，还应考虑文件锁定或其他同步方法。

二进制数据操作

pwrite 擅长二进制文件操作。此示例显示了在二进制文件中修改特定字节，同时保持其他字节不变。

binary_edit.py

import os

# Create a binary file with a pattern
with open("data.bin", "wb") as f:
    f.write(bytes(range(256)))  # 0x00 to 0xFF

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

# Modify specific bytes
os.pwrite(fd, b"\xFF\xFF", 0x10)  # Overwrite positions 0x10-0x11
os.pwrite(fd, b"\xAA\xBB", 0x80)  # Overwrite positions 0x80-0x81

# Verify changes
os.lseek(fd, 0, os.SEEK_SET)
content = os.read(fd, 256)

print(f"Byte at 0x10: {hex(content[0x10])}")
print(f"Byte at 0x11: {hex(content[0x11])}")
print(f"Byte at 0x80: {hex(content[0x80])}")
print(f"Byte at 0x81: {hex(content[0x81])}")

os.close(fd)

这会创建一个具有已知模式的二进制文件，然后使用 pwrite 修改特定的字节范围。文件的其余部分保持不变。

此技术对于二进制文件格式很有用，在这种格式中，您需要修改较大文件中的特定标头或数据结构。

性能比较

此示例比较了在对大文件进行随机访问时 pwrite 与常规 write 的性能。

performance.py

import os
import time
import random

def test_pwrite(fd, positions):
    start = time.time()
    for pos in positions:
        os.pwrite(fd, b"X", pos)
    return time.time() - start

def test_write(fd, positions):
    start = time.time()
    for pos in positions:
        os.lseek(fd, pos, os.SEEK_SET)
        os.write(fd, b"X")
    return time.time() - start

# Create a large test file
fd = os.open("perf_test.dat", os.O_RDWR | os.O_CREAT)
os.ftruncate(fd, 1000000)  # 1MB file

# Generate random positions
positions = [random.randint(0, 999999) for _ in range(1000)]

# Test pwrite
pwrite_time = test_pwrite(fd, positions)

# Test regular write
write_time = test_write(fd, positions)

os.close(fd)

print(f"pwrite time: {pwrite_time:.4f} seconds")
print(f"write time: {write_time:.4f} seconds")
print(f"Ratio: {write_time/pwrite_time:.2f}x")

这会基准测试写入到 1MB 文件中的 1000 个随机位置。pwrite 通常通过避免重复的查找操作来获得更好的性能。

性能差异随着更多随机写入而增长，这使得 pwrite 非常适合某些类数据库操作。

安全注意事项

文件描述符： 需要正确的文件描述符管理
原子写入： 最大 PIPE_BUF 字节的写入是原子的
错误处理： 始终检查返回值以进行部分写入
并发： 对于多线程使用是安全的，但需要适当的协调
平台支持： 在类 Unix 系统上可用，不在 Windows 上

最佳实践

用于随机访问： 非常适合在已知偏移量处写入
检查返回值： 验证是否写入了所有字节
与 pread 结合使用： 用于读-修改-写循环
考虑文件同步： 在需要持久性时使用 fsync
更喜欢用于大文件： 比 seek+write 更有效

资料来源

作者

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

列出所有 Python 教程。