在 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⚠️ 注意事项：
本文档主要讲述的是Matlab语言的特点；Matlab具有用法简单、灵活、程式结构性强、延展性好等优点，已经逐渐成为科技计算、视图交互系统和程序中的首选语言工具。特别是它在线性代数、数理统计、自动控制、数字信号处理、动态系统仿真等方面表现突出，已经成为科研工作人员和工程技术人员进行科学研究和生产实践的有利武器。希望本文档会给有需要的朋友带来帮助；感兴趣的朋友可以过来看看
- 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免费学习笔记（深入）”；
