结论是：用Composer构建流式数据处理框架，核心在于选对底层驱动（如react/stream或swoole）、隔离阻塞操作、严格控制数据chunk生命周期；盲目require各类parser/transformer库易因同步I/O或非协程安全导致流水线卡死。
直接说结论：用 Composer 构建流式数据处理框架，核心不是“装一堆包”，而是选对底层驱动（如 react/stream 或 swoole）、隔离阻塞操作、并严格控制数据 chunk 的生命周期。盲目 require 各种 parser 或 transformer 库，反而会因同步 I/O 或非协程安全设计导致整个流水线卡死。
react/stream 不能直接当“流处理器”用
React\Stream\ReadableResourceStream 只是把 fread() 封装进事件循环，chunk 大小由系统 buffer 决定，无法控制——这对 XML/JSON/Protobuf 等需结构化解析的场景极不友好。比如读到半截 JSON 对象或未闭合的 XML 标签，on('data') 就会吐出非法片段。
- 必须自己封装循环：
fread($fp, 8192)+ 手动$stream->emit('data', $buf)，才能保证最小可控粒度 - 遇到换行分隔的日志文件，得在
on('data')里做缓冲拼接，直到拿到完整行再 emit，否则下游解析器会报SyntaxError: Unexpected token -
ReadableResourceStream不处理编码转换（如 GBK → UTF-8），需在 emit 前用mb_convert_encoding()显式转码，否则中文字段直接乱码
orchestra/parser 和 protobuf-php/protobuf 的流式适配要点
这两个库本身是同步解析器，直接喂 chunk 会失败。必须等数据攒够一个完整单元（如一个 XML 根节点闭合、一个 Protobuf message 完整）再调用解析方法。
- 对 XML：用
xml_parser_create()配合XML_Parse()增量解析，监听XML_ELEMENT_START和XML_ELEMENT_END事件，只在XML_ELEMENT_END且深度归零时触发转换逻辑 - 对 Protobuf：先用
file_get_contents()读取完整 message（需提前知道长度前缀），或用unpack('N', fread($fp, 4))解出 length 字段，再读对应字节数——protobuf-php/protobuf的mergeFromString()不接受部分二进制 - 别在
on('data')里直接 newOrchestra\Parser\Xml\XmlParser，它内部会重置 parser state，导致跨 chunk 解析失败
Hyperf 协程环境下的流式陷阱
在 Hyperf 中用 react/stream 会破坏协程上下文，因为 React 是基于 PHP stream_select() 的回调模型，与 Swoole 协程调度器不兼容。结果就是：CPU 占用飙升、超时异常频发、go() 启动的协程无法 await 流事件。
- 正确路径是放弃 React，改用
hyperf/utils提供的Coroutine\Channel+co::readFile()分块读取，再用Channel::push()向下游投递 - 若必须对接外部流式服务（如 Kafka），用
hyperf/kafka客户端，它内部已做协程封装，支持yield $consumer->consume() -
composer require lizhichao/one-sm这类国密库虽支持流式加解密，但其sm4_encrypt_ecb()方法要求输入长度为 16 字节倍数——必须在分块前做 padding，且 padding 方式要和下游系统一致，否则解密失败
真正难的不是怎么读，是怎么界定“一个完整数据单元”。XML 要靠标签栈深度，Protobuf 要靠长度前缀，日志要靠换行符，CSV 要看引号和逗号嵌套。这些边界判断逻辑一旦写错，整个流就不可恢复——没有“重试”概念，只有丢弃或阻塞。所以第一行代码不该是 composer require，而是白板上画清楚你的数据协议格式。
