在 linux 下使用 `os.pipe()` 时，由于管道文件描述符不支持随机寻址（`lseek`），无法直接调用 `tell()` 获取累计写入字节数；但可通过封装一个具备 `write()` 和 `flush()` 方法的类，将计数逻辑内置于写入过程，从而准确追踪总字节数。
os.pipe() 返回一对底层文件描述符（read_fd, write_fd），它们本质上是不可寻址的流式通道，因此任何依赖 lseek 的操作（如 fileobj.tell()）都会抛出 IOError: [Errno 29] Illegal seek。标准库中没有内置的“带计数的管道包装器”，但 Python 的 I/O 协议非常灵活：只要对象实现了 write()（接受 bytes/str）和 flush() 方法，就可被多数 I/O 相关函数（如 print(…, file=…)、sys.stdout = … 或 subprocess.Popen(stdout=…)）接受。
以下是一个轻量、线程安全（可选加锁）、符合 Python I/O 协议的累计计数写入器示例：
import os
import threading
class CountingWriter:
def __init__(self, fd, encoding='utf-8'):
self.fd = fd
self._count = 0
self._lock = threading.Lock()
self.encoding = encoding
def write(self, data):
if isinstance(data, str):
data = data.encode(self.encoding)
# 原子写入并累加
with self._lock:
n = os.write(self.fd, data)
self._count += n
return n
def flush(self):
# 管道无缓冲，flush 可为空，但需存在以满足协议
pass
@property
def written_bytes(self):
with self._lock:
return self._count
# 使用示例
rfd, wfd = os.pipe()
# 将计数写入器绑定到写端
writer = CountingWriter(wfd)
# 模拟写入
writer.write(b"Hello")
writer.write(" 世界\n".encode('utf-8'))
writer.flush()
print(f"累计写入字节数: {writer.written_bytes}") # 输出: 13（"Hello" + " 世界\n" = 5 + 8）
# 注意：读端仍需用 os.read(rfd, ...) 手动读取
# os.close(rfd); os.close(wfd) —— 实际使用中请记得关闭 fd
⚠️ 注意事项：
