PHP stream_select 怎么同时监听多个子进程管道

PHP 同时读取多个子进程输出时，不要把某一条管道写成阻塞式的“读完再读下一条”。更稳妥的做法是用 proc_open() 保存每个子进程的 stdout 管道，把当前仍然活跃的管道复制到 stream_select()，只读取本轮已经就绪的流；管道到达 EOF 后立即关闭并从主表移除。

关键点只有一个：stream_select() 会原地改写传入的 read 数组，所以主监听表不能直接复用，必须每轮重新构造临时数组。

官方地址：https://www.php.net/manual/en/function.stream-select.php

下面用两个短命令模拟两个子进程，示例只聚焦 PHP CLI 下的 stdout 管道、超时和资源回收。

先把多个子进程变成可管理的管道集合

proc_open() 会把描述符数组里的管道端返回到 $pipes。父进程要保存三类信息：进程句柄、可读的 stdout 管道，以及管道对应的任务名。这样读到一段文本时，才能知道它来自哪个子进程。

下面的 spawn() 只打开 stdout 读管道；子进程的 stderr 在真实项目中也应单独处理，避免错误输出堵住进程。

 ['pipe', 'r'], // 父进程不向 stdin 写数据。
        1 => ['pipe', 'w'], // 父进程读取子进程 stdout。
        2 => ['pipe', 'w'], // 单独保留 stderr，生产环境应持续消费它。
    ];
    $pipes = [];
    $process = proc_open($command, $descriptorSpec, $pipes);
    if (!is_resource($process)) {
        throw new RuntimeException("无法启动 {$name}");
    }

    // 不需要输入时立即关闭 stdin，避免子进程等待 EOF。
    fclose($pipes[0]);
    stream_set_blocking($pipes[1], false); // 读取时不因单条管道卡住整个监控器。
    stream_set_blocking($pipes[2], false); // stderr 也切到非阻塞，便于后续扩展。
    return ['name' => $name, 'process' => $process, 'stdout' => $pipes[1], 'stderr' => $pipes[2]];
}

$jobs = [spawn('任务 A', 180), spawn('任务 B', 420)];
$active = [];
foreach ($jobs as $job) {
    // 用资源 ID 作为主键，避免不同管道对象混在一起。
    $active[(int) $job['stdout']] = $job;
}
?>

监听循环为什么要重建 read 数组

stream_select($read, ...) 返回后，$read 只保留已经就绪或已经到 EOF 的流，原数组会被修改。若把它当成永久监听表，下一轮就可能漏掉仍在工作的管道。因此主表保存在 $active，每轮用 array_values() 生成新的 read 数组。

 $job['stdout'], array_values($active));
    $write = null;
    $except = null;

    // 2 秒是上限，不代表一定要等满；有数据时会提前返回。
    $ready = stream_select($read, $write, $except, 2, 0);
    if ($ready === false) {
        throw new RuntimeException('stream_select 被中断或参数无效');
    }
    if ($ready === 0) {
        // 超时可以在这里做心跳、记录进度或检查取消信号。
        continue;
    }

    foreach ($read as $stream) {
        $key = (int) $stream;
        $job = $active[$key] ?? null;
        if ($job === null) {
            continue; // 管道可能在上一轮清理，避免访问已删除的状态。
        }

        $chunk = fread($stream, 8192);
        if ($chunk !== false && $chunk !== '') {
            echo '[' . $job['name'] . '] ' . rtrim($chunk) . PHP_EOL;
        }
        if (feof($stream)) {
            // EOF 表示子进程 stdout 已结束，先关管道，再从主表移除。
            fclose($stream);
            fclose($job['stderr']);
            proc_close($job['process']);
            unset($active[$key]);
        }
    }
}
?>

这里使用 === false 判断失败，因为返回值为 0 代表正常超时，不是错误。读数组中的流也可能因为 EOF 被报告为就绪，此时 fread() 得到空字符串并不意味着要继续永久监听，而是应该执行关闭和移除。

超时、读取和清理要分别处理

秒和微秒共同组成等待上限。2, 0 适合示例；如果希望由事件唤醒，可以把秒参数设为 null，但循环里就不能依赖固定时间做心跳。把两个参数都设为零会变成忙轮询，官方手册也提醒这种写法可能持续消耗 CPU。

如果业务只关心标准输出，stderr 也不能永远无人消费。可以把 stderr 放入另一张监听表，或者在子进程配置中把它重定向到文件；关键是保证每个资源都有明确的关闭路径。

常见问题

为什么不直接对每条管道调用 fgets？

阻塞式读取一条管道时，父进程可能一直等不到它的数据，其他已经有输出的子进程也无法及时处理。先用 stream_select() 找出可读流，才能把等待范围限制在“任意一条管道有变化”。

stream_select 的 read 数组能不能复用？

不能把它当作永久列表。函数会按本次结果原地筛选数组；正确做法是保存自己的 $active，每轮重新映射出 $read，清理 EOF 后再进入下一轮。

总结：多个子进程管道的核心不是把读取写成更复杂，而是把“活跃管道主表”和“本轮就绪副本”分开。再分别处理 false、超时、可读数据和 EOF，stream_select() 就能承担一个轻量的多管道等待器。