在 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⚠️ 注意事项：
- CountingWriter 不提供 read() 或 close()，它仅代理写操作；管道生命周期仍需由调用方管理；
- 若用于 sys.stdout = writer 场景，需确保所有输出均经此对象（例如 print("msg", file=writer)），且避免混用原生 os.write()；
- 多线程写入时务必加锁（如上例），否则 _count 可能出现竞态；
- 对于 subprocess.Popen(stdout=writer)，需注意子进程会继承该 fd，此时 writer 应仅作为“写入目标”而非主动读取方。
总结：虽然 os.pipe() 本身不提供字节计数能力，但借助 Python 的鸭子类型特性，自定义一个兼容 io.TextIOBase/io.BufferedIOBase 协议的写入器，即可在零侵入前提下精准统计累计写入量——这是兼顾简洁性与可扩展性的典型 Pythonic 解决方案。
立即学习“Python免费学习笔记（深入）”；
