如何监听PHP流数据传输，排查Laravel流场景下的慢查询

你需要监控流传输时数据块的发送间隔，定位因慢查询导致间隔超1分钟的端点，以下是无需在业务代码中逐个修改的解决方案：

方案一：自定义PHP流包装器监控输出流 #

通过自定义流包装器拦截php://output的写入操作，自动记录每次数据发送的时间戳并计算间隔：

1. 创建流监控类

<?php

namespace App\Helpers;

class StreamMonitor
{
    private $stream;
    private $lastWriteTime;
    private $requestPath;

    public function __construct()
    {
        $this->lastWriteTime = microtime(true);
        $this->requestPath = request()->path();
    }

    public function stream_open($path, $mode, $options, &$opened_path)
    {
        $this->stream = fopen($path, $mode);
        return $this->stream !== false;
    }

    public function stream_write($data)
    {
        $currentTime = microtime(true);
        $interval = $currentTime - $this->lastWriteTime;

        // 记录间隔超过60秒的情况
        if ($interval > 60) {
            logger()->warning('流数据块间隔过长', [
                'request_path' => $this->requestPath,
                'interval_seconds' => round($interval, 2),
                'timestamp' => date('Y-m-d H:i:s', $currentTime)
            ]);
        }

        $this->lastWriteTime = $currentTime;
        return fwrite($this->stream, $data);
    }

    public function stream_close()
    {
        return fclose($this->stream);
    }

    // 实现其他必要的流方法
    public function stream_read($count) { return fread($this->stream, $count); }
    public function stream_eof() { return feof($this->stream); }
    public function stream_stat() { return fstat($this->stream); }
}

2. 在服务提供者中注册包装器
   在AppServiceProvider的boot方法中添加：

if (!in_array('monitor', stream_get_wrappers())) {
    stream_wrapper_register('monitor', \App\Helpers\StreamMonitor::class);
}

3. 统一替换流打开方式
   将业务代码中所有fopen('php://output', 'w')替换为fopen('monitor://php://output', 'w')，如果想彻底避免修改业务代码，可以结合Laravel响应宏统一处理。

方案二：Laravel全局中间件包装流响应 #

通过中间件拦截所有StreamResponse，自动包装其回调函数添加监控逻辑，完全无需修改业务代码：

1. 创建中间件

<?php

namespace App\Http\Middleware;

use Closure;
use Illuminate\Http\StreamedResponse;

class MonitorStreamResponse
{
    public function handle($request, Closure $next)
    {
        $response = $next($request);

        if ($response instanceof StreamedResponse) {
            $originalCallback = $response->getCallback();
            $lastWriteTime = microtime(true);
            $requestPath = $request->path();

            $wrappedCallback = function () use ($originalCallback, &$lastWriteTime, $requestPath) {
                // 重写fputcsv函数实现监控
                $originalFputcsv = 'fputcsv';
                $GLOBALS['fputcsv'] = function ($stream, $fields) use (&$lastWriteTime, $requestPath) {
                    $currentTime = microtime(true);
                    $interval = $currentTime - $lastWriteTime;

                    if ($interval > 60) {
                        logger()->warning('流数据块间隔过长', [
                            'request_path' => $requestPath,
                            'interval_seconds' => round($interval, 2),
                            'timestamp' => date('Y-m-d H:i:s', $currentTime)
                        ]);
                    }

                    $lastWriteTime = $currentTime;
                    return $originalFputcsv($stream, $fields);
                };

                // 执行原回调逻辑
                $originalCallback();

                // 恢复原函数
                $GLOBALS['fputcsv'] = $originalFputcsv;
            };

            $response->setCallback($wrappedCallback);
        }

        return $response;
    }
}

2. 注册全局中间件
   在app/Http/Kernel.php的$middleware数组中添加：

\App\Http\Middleware\MonitorStreamResponse::class,

关键注意事项 #

• 日志中务必记录请求路径，方便快速定位具体端点
• 可扩展监控逻辑，添加数据块大小、总耗时等维度的记录
• 高流量场景建议添加日志采样机制，避免日志量过大
• 测试时可在chunk回调中加入sleep(61)模拟慢查询，验证监控是否生效

内容的提问来源于stack exchange，提问作者Chris