如何通过stream_get_contents()连续监听多个TCP流？

Question

默认情况下stream_get_contents 等待并侦听 60 秒。 如果我有多个流并且想要连续收听所有流。

在 foreach 中，如果我正在收听一个流，我就无法收听其他流。

连续监听和捕获所有流的流输出的解决方案是什么？

while(true){
   //$streamArray is an array of streams obtained by stream_socket_client("tcp://..);
   foreach($streamArray as $stream){
        fputs($stream,$command);
        stream_get_contents($stream);  // and update file/DB (Blocking call)
        }
    }

注意：对于我已经完成的每个流stream_set_blocking( $stream , true );

更新：

我的要求是在一段时间内收听所有流媒体，比如 30 分钟。同时我不能听 2 流。如果我有 5 个流，我的代码就是 time division multiplexing，在 30 分钟内，每个单独的流将只记录 6 分钟

我有一种解决方案，可以对单个流发出 AJAX 请求并独立记录。 当然我不想做这种多重 AJAX 调用方法，因为它会导致更多的代码以及更多的 CPU。

Answer 1

使用stream_set_blocking($resource, true)，您开始同步读取流，这意味着每次调用fread() 都会等待直到有数据要读取。然后调用stream_get_contents()，它从阻塞流中读取直到到达EOF（关闭）。结果是您一个接一个地读取流，而不是“同时”读取。

以这种方式读取流很常见并且最容易编码，但是当您想同时处理多个流时，您必须自己处理缓冲区、计时和“流结束”检测。使用您当前的代码，这部分是通过阻塞流和stream_get_contents() 抽象出来的。

要同时读取多个流，您应该重写代码以异步读取流。

// $streamArray is an array of streams obtained by stream_socket_client("..");

$buffers = array();
$num_streams = 0;
foreach($streamArray as $stream) {
    // set streams non-blocking for asynchroneous reading
    stream_set_blocking($stream, false);
    // write command to all streams - multiplexing
    fputs($stream, $command);
    // initialize buffers
    $buffers[$stream] = '';
    $num_streams++;
}
while($num_streams > 0) {
    // use stream_select() to wait for new data and not use CPU at 100%.
    // note that the stream arrays are passed by reference and are modified
    // by stream_select() to reflect which streams are modified / have a
    // new event. Because of this we use a copy of the original stream array.
    // also note that due to a limitation in ZE you can not pass NULL directly
    // for more info: read the manual    https://php.net/stream_select
    $no_stream = NULL;
    $select_read = $streamArray;
    stream_select($select_read, $no_stream, $no_stream, null);

    // if there is new data, read into the buffer(s)
    foreach($select_read as $stream) {
        $buffers[$stream] .= fread($stream, 4096);
        // check if the stream reached end
        if (feof($stream)) {
            $key = array_search($stream, $streamArray);
            // close stream properly
            $num_streams--;
            fclose($stream);
            // remove stream from array of open streams
            unset($streamArray[$key]);
        }
    }
}
// do something with your buffers - our use them inside the while() loop already
print_r($buffers);

Answer 2

PHP 在服务器上作为单个进程运行。执行此类操作的“最佳”方法是为每个流生成一个线程。每个线程都会有一个单独的流来收听。多线程自动利用多个 CPU 并允许它们并行运行。

不幸的是，PHP不允许允许您创建线程。但是，有几种变通方法可供选择（注意：this question 可能对您有所帮助）。

选项 1：克隆进程

PHP 确实 允许您使用pcntl_fork 之类的函数克隆当前进程。基本上，您可以创建当前 PHP 进程的副本，该进程将接管并开始运行。只需对需要收听的每个流执行此操作即可。复杂之处在于确保每个进程都监听正确的流。 （提示：有一个流列表，并为列表中的下一个流使用一个标志。然后执行 fork [克隆进程]。子进程将开始监听该流。但是，在父级中，将标志更新为列表中的下一个流，并创建另一个子级，等等......）

选项 2：多个 PHP 实例

您可以使用 Cronjob、多个调用或其他一些服务（甚至是命令行调用）来调用每个 PHP 脚本，或者将参数传递给脚本以识别要监听的流。这将并行运行每个 PHP 脚本/文件，并实现您想要的。

选项 3：不要使用 PHP

PHP 在技术上并不是为这样的东西而构建的，而这很可能在 C、C++ 或 Java 中很容易实现。您应该重新考虑您的设置。

选项 4：创建 PHP 或 Apache 模块

您可以用 C 语言为 PHP 或 Apache 编写一个模块，然后在 PHP 中使用您的函数调用。这可能会很复杂，我不推荐它。

总而言之，我建议重新考虑您的设置，而不是使用 PHP。但是，如果您仅限于 PHP，则可以使用多个 AJAX 调用。

Answer 3

连续监听和捕获所有流的流输出的解决方案是什么？

This issue 困扰程序员很长时间了。如果您想找到合适的解决方案来解决您的问题，一本由 W. Richard Stevens 撰写的名为 UNIX Network Programming 的书将是一个很好的起点。