Guzzle HTTP 流处理：如何在不阻塞于read()方法的情况下实时读取数据？

这个场景我之前做长链接推送的时候踩过一模一样的坑！Guzzle的read()方法确实在处理不定长、带结束标记的流数据时有点反直觉，尤其是需要实时处理的场景，稍不注意就会卡在那里无限等待。

核心问题就是Guzzle的read()是基于流的字节读取，它不管你的消息边界——如果你指定的读取长度大于当前流中已有的数据，它会阻塞直到拿到足够的字节；如果小于一条完整消息的长度，就会把剩余的消息片段留在流里，而你的代码如果没处理这些片段，下一次就会等着凑够数据，直接僵住。

下面是我当时摸索出来的可行方案，亲测能用：

1. 手动管理缓冲区+按结束标记拆分完整消息 #

自己接管数据的拼接和拆分，不管每次读多少字节，都把读取到的内容暂存在缓冲区里，每次检查缓冲区里是否出现了你的消息结束标记，一旦找到就提取完整消息，剩下的内容留在缓冲区等待下一次拼接。

示例代码：

// 发起长链接请求，开启流模式
$client = new \GuzzleHttp\Client();
$response = $client->request('GET', 'your-long-push-endpoint', [
    'stream' => true,
    'timeout' => 0, // 禁用超时，或者根据需求设置一个合理的长超时
]);
$stream = $response->getBody();

$buffer = '';
$endMarker = '你的消息结束标记，比如特定字符串、\n或者\r\n';
$markerLength = strlen($endMarker);

while (!$stream->eof()) {
    // 每次读取1024字节，平衡读取性能和实时性，可根据你的消息大小调整
    $chunk = $stream->read(1024);
    if ($chunk === false) {
        break; // 流已关闭，退出循环
    }

    // 把新读取的块追加到缓冲区
    $buffer .= $chunk;

    // 循环检查缓冲区里的结束标记（可能一次读取到多条消息）
    while (($markerPos = strpos($buffer, $endMarker)) !== false) {
        // 提取标记之前的完整消息
        $fullMessage = substr($buffer, 0, $markerPos);
        // 这里处理你的消息，比如推送给前端客户端
        $this->handleReceivedMessage($fullMessage);

        // 更新缓冲区：保留标记之后的剩余内容，用于下一次拼接
        $buffer = substr($buffer, $markerPos + $markerLength);
    }
}

// 最后记得关闭流
$stream->close();

2. 非阻塞模式优化：避免无意义的阻塞等待 #

如果你的服务端是间隔推送消息（不是持续发数据），阻塞模式下read()会一直等着新数据，导致进程完全被占用，没法处理其他逻辑。这时候可以把流设置为非阻塞模式，配合短暂休眠来平衡实时性和CPU消耗。

在上面的代码基础上，添加非阻塞设置：

// 获取底层的流资源，设置为非阻塞模式
$streamResource = $stream->detach();
stream_set_blocking($streamResource, false);
// 把流资源重新attach回Guzzle的流对象
$stream->attach($streamResource);

// 调整循环逻辑
while (true) {
    $chunk = $stream->read(1024);
    if ($chunk === false) {
        // 非阻塞模式下返回false表示当前没有数据，短暂休眠避免CPU空转
        usleep(10000); // 休眠10毫秒，可根据需求调整
        continue;
    }
    if (empty($chunk)) {
        break; // 流已关闭，退出循环
    }

    // 下面的缓冲区处理和消息提取逻辑和之前一致
    $buffer .= $chunk;
    while (($markerPos = strpos($buffer, $endMarker)) !== false) {
        $fullMessage = substr($buffer, 0, $markerPos);
        $this->handleReceivedMessage($fullMessage);
        $buffer = substr($buffer, $markerPos + $markerLength);
    }
}

几个关键注意事项 #

• 结束标记必须唯一：一定要确保你的消息正文里不会出现和结束标记完全相同的内容，否则会错误拆分消息。如果无法避免这个情况，可以和服务端协商改用「长度前缀+消息内容」的协议格式，先读取固定长度的长度值，再读取对应长度的消息内容。
• 连接保活机制：长链接很容易因为网络波动静默断开，建议和服务端约定心跳包（比如每隔30秒发送一个特定的心跳消息），如果客户端长时间没收到心跳，就主动断开重连。
• 超时设置：如果禁用了timeout=0，一定要有其他的连接健康检测逻辑，不然进程可能会一直挂在无效的连接上。

备注：内容来源于stack exchange，提问作者user27909701