Workerman 本身不内置流式清洗能力，而是作为高性能接入层将大数据流拆包后交由外部组件清洗；不能在 onMessage 中直接遍历大数组或加载大文件，否则会导致进程阻塞和内存溢出；应使用 Generator 分块读取、Redis 队列解耦、长度前缀协议处理粘包，确保数据不丢不错不卡。
Workerman 本身不内置流式清洗能力，但能作为高性能 TCP/WebSocket 接入层 + 协调层，把大数据流拆成可控小包、交由外部组件（如 Redis 队列、协程处理器、生成器）完成清洗。关键不是“Workerman 做清洗”，而是它怎么把脏数据稳稳送进去、把干净结果准准推出来。
为什么不能在 onMessage 里直接 foreach 大数组？
Workerman 的 onMessage 回调是同步执行的，一旦你在里面做 array_map、json_decode(file_get_contents($big_file)) 或遍历百万行 CSV，整个 worker 进程会卡住，后续连接全部排队——这不是高并发，是高阻塞。
- 单次处理超过 10MB JSON 或 5 万行 CSV，基本触发 PHP 内存溢出（尤其没设
memory_limit） - 用
file_get_contents加载大文件，会一次性占满内存，且无法中断或进度反馈 - 多进程下各 worker 独立运行，没有共享状态，无法协作分片清洗
用 Generator + 分块读取替代全量加载
清洗逻辑必须脱离 onMessage 主线程，改用 PHP 原生 Generator 流式解析。比如处理上传的 CSV 流：
function csvLineGenerator($handle) {
while (($line = fgetcsv($handle)) !== false) {
yield $line;
}
}
// 在异步任务中调用（非 onMessage 直接调用）
$fp = fopen('php://input', 'r');
foreach (csvLineGenerator($fp) as $row) {
// 每行校验、转换、过滤，再 push 到 Redis 或写入临时文件
$cleaned = cleanRow($row);
$redis->lPush('clean_queue', json_encode($cleaned));
}
- 必须用
fopen('php://input', 'r')而非$_FILES，避免 PHP 自动缓存上传文件到临时目录再读取 -
Generator不会把全部数据 load 进内存，只保留当前行上下文 - 清洗后的结果别直接 echo 或 send，走队列解耦，防止下游慢拖垮入口
如何让多个 Workerman 进程协同清洗同一批数据？
Workerman 没有原生分布式任务调度，得靠外部协调。典型做法是：用 Redis List 做任务队列，配合 Timer::add() 轮询消费：
- 入口
onMessage只做一件事：接收原始数据包 → 拆成固定大小（如 4KB）→ 加包头（含总长度、序号）→$redis->lPush('raw_chunks', $packet) - 另起一个独立的 BusinessWorker（或 CLI 脚本），用
Timer::add(0.1, function() { $chunk = $redis->rPop('raw_chunks'); if ($chunk) processChunk($chunk); }) - 清洗完的数据写入另一个队列
clean_results，再由 WebSocket worker 广播给前端
注意：rPop 是原子操作，天然支持多进程并发消费，无需加锁。
TCP 粘包/拆包不处理，清洗就一定错乱
如果走的是自定义 TCP 协议（非 WebSocket），onMessage 收到的数据大概率是半包或粘包。你按换行或逗号切分，第一行可能缺字段，最后一行可能跨包——清洗规则全崩。
- 必须实现长度前缀协议：每个包开头 4 字节 int 存 payload 长度，接收端用
$connection->recv(4)先读长度，再读对应字节数 - 不要依赖
\n或\0做分隔符，网络传输中这些字符可能出现在业务数据里 - 推荐分包大小设为
4096～8192字节，太小增加包头开销，太大仍可能触发内存警戒线
真正难的从来不是“怎么写清洗逻辑”，而是怎么让数据在进、传、出三个环节都不丢、不错、不卡——Workerman 只管前两环的稳定性，第三环得你亲手串起来。
