Workerman 本身不内置流式清洗能力，而是作为高性能接入层将大数据流拆包后交由外部组件清洗；不能在 onMessage 中直接遍历大数组或加载大文件，否则会导致进程阻塞和内存溢出；应使用 Generator 分块读取、Redis 队列解耦、长度前缀协议处理粘包，确保数据不丢不错不卡。
Workerman 本身不内置流式清洗能力，但能作为高性能 TCP/WebSocket 接入层 + 协调层，把
大数据
流拆成可控小包、交由外部组件（如 Redis 队列、协程处理器、生成器）完成清洗。关键不是“Workerman 做清洗”，而是它怎么把脏数据稳稳送进去、把干净结果准准推出来。
为什么不能在 onMessage 里直接 foreach 大数组？
Workerman 的
回调是同步执行的，一旦你在里面做
、
或遍历百万行 CSV，整个 worker 进程会卡住，后续连接全部排队——这不是高并发，是高阻塞。
单次处理超过 10MB JSON 或 5 万行 CSV，基本触发 PHP 内存溢出（尤其没设
）
用
加载大文件，会一次性占满内存，且无法中断或进度反馈
多进程下各 worker 独立运行，没有共享状态，无法协作分片清洗
用 Generator + 分块读取替代全量加载
清洗逻辑必须脱离
主线程，改用 PHP 原生
流式解析。比如处理上传的 CSV 流：
必须用
而非
，避免 PHP 自动缓存上传文件到临时目录再读取
不会把全部数据 load 进内存，只保留当前行上下文
清洗后的结果别直接 echo 或 send，走队列解耦，防止下游慢拖垮入口
如何让多个 Workerman 进程协同清洗同一批数据？
Workerman 没有原生分布式任务调度，得靠外部协调。典型做法是：用 Redis List 做任务队列，配合
轮询消费：
用Apache Spark进行大数据处理
本文档主要讲述的是用Apache Spark进行大数据处理——第一部分：入门介绍；Apache Spark是一个围绕速度、易用性和复杂分析构建的大数据处理框架。最初在2009年由加州大学伯克利分校的AMPLab开发，并于2010年成为Apache的开源项目之一。 在这个Apache Spark文章系列的第一部分中，我们将了解到什么是Spark，它与典型的MapReduce解决方案的比较以及它如何为大数据处理提供了一套完整的工具。希望本文档会给有需要的朋友带来帮助；感
下载
入口
只做一件事：接收原始数据包 → 拆成固定大小（如 4KB）→ 加包头（含总长度、序号）→
另起一个独立的 BusinessWorker（或 CLI 脚本），用
清洗完的数据写入另一个队列
，再由 WebSocket worker 广播给前端
注意：
是原子操作，天然支持多进程并发消费，无需加锁。
TCP 粘包/拆包不处理，清洗就一定错乱
如果走的是自定义 TCP 协议（非 WebSocket），
收到的数据大概率是半包或粘包。你按换行或逗号切分，第一行可能缺字段，最后一行可能跨包——清洗规则全崩。
必须实现长度前缀协议：每个包开头 4 字节 int 存 payload 长度，接收端用
先读长度，再读对应字节数
不要依赖
或
做分隔符，网络传输中这些字符可能出现在业务数据里
推荐分包大小设为
～
字节，太小增加包头开销，太大仍可能触发内存警戒线
真正难的从来不是“怎么写清洗逻辑”，而是怎么让数据在进、传、出三个环节都不丢、不错、不卡——Workerman 只管前两环的稳定性，第三环得你亲手串起来。
onMessagearray_mapjson_decode(file_get_contents($big_file))memory_limitfile_get_contentsonMessageGeneratorfunction 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')$_FILESGeneratorTimer::add()onMessage$redis->lPush('raw_chunks', $packet)Timer::add(0.1, function() { $chunk = $redis->rPop('raw_chunks'); if ($chunk) processChunk($chunk); })clean_resultsrPoponMessage$connection->recv(4)\n\040968192