如何编写CLI可定时运行的PHP脚本将SSE Stream数据存入MySQL

核心逻辑说明 #

不需要依赖浏览器环境，直接通过cURL拉取SSE流式响应，解析后直接写入MySQL，脚本内置超时控制避免无限挂起，搭配crontab实现10分钟定时执行。

完整脚本代码 #

<?php
// 配置项
define('SSE_URL', 'https://sseAPIUrl.com');
define('MYSQL_HOST', '你的MySQL地址');
define('MYSQL_DB', '你的库名');
define('MYSQL_USER', '你的用户名');
define('MYSQL_PASS', '你的密码');
define('MYSQL_TABLE', '你的表名');
define('MAX_RUN_TIME', 300); // 单次脚本最长运行时间（秒），这里设5分钟，避免和下一次定时任务重叠
define('LOCK_FILE', __DIR__ . '/sse_collect.lock');

// 加独占锁，防止多个脚本同时运行
$lockFp = fopen(LOCK_FILE, 'w');
if (!$lockFp || !flock($lockFp, LOCK_EX | LOCK_NB)) {
    die(date('[Y-m-d H:i:s]') . " 已有脚本在运行，退出\n");
}

// 初始化PDO连接
try {
    $pdo = new PDO(
        "mysql:host=" . MYSQL_HOST . ";dbname=" . MYSQL_DB . ";charset=utf8mb4",
        MYSQL_USER,
        MYSQL_PASS,
        [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]
    );
    $insertStmt = $pdo->prepare("INSERT INTO " . MYSQL_TABLE . " (content, create_time) VALUES (:content, NOW())");
} catch (PDOException $e) {
    die(date('[Y-m-d H:i:s]') . " 数据库连接失败：" . $e->getMessage() . "\n");
}

// SSE消息解析缓冲区
$buffer = '';

// 初始化cURL
$ch = curl_init(SSE_URL);
curl_setopt_array($ch, [
    CURLOPT_FOLLOWLOCATION => true,
    CURLOPT_SSL_VERIFYPEER => false, // 如果SSE地址是自签SSL可以开，正规证书建议关闭该配置
    CURLOPT_TIMEOUT => MAX_RUN_TIME,
    CURLOPT_WRITEFUNCTION => function($ch, $data) use (&$buffer, $insertStmt) {
        $buffer .= $data;
        // 按SSE协议的空行分隔消息
        while (($pos = strpos($buffer, "\n\n")) !== false) {
            $message = substr($buffer, 0, $pos);
            $buffer = substr($buffer, $pos + 2);
            
            // 解析data字段
            if (preg_match('/^data: (.*)$/m', $message, $matches)) {
                $streamData = $matches[1];
                // 写入数据库
                try {
                    $insertStmt->execute([':content' => $streamData]);
                    echo date('[Y-m-d H:i:s]') . " 写入成功，数据：" . mb_substr($streamData, 0, 50) . "...\n";
                } catch (PDOException $e) {
                    echo date('[Y-m-d H:i:s]') . " 写入失败：" . $e->getMessage() . "\n";
                }
            }
        }
        return strlen($data);
    }
]);

// 执行请求
curl_exec($ch);

// 收尾处理
if (curl_errno($ch)) {
    echo date('[Y-m-d H:i:s]') . " cURL错误：" . curl_error($ch) . "\n";
}
curl_close($ch);
flock($lockFp, LOCK_UN);
fclose($lockFp);
unlink(LOCK_FILE);
?>

定时任务配置 #

Linux环境下直接配置crontab即可，执行crontab -e后添加以下规则：

*/10 * * * * /usr/bin/php /你脚本的实际路径/sse_collect.php >> /你要存日志的路径/sse_collect.log 2>&1

注意把PHP路径、脚本路径、日志路径替换成你服务器的实际路径，日志目录要保证有写入权限。

注意事项 #

• 如果你的SSE返回的是JSON格式数据，可以在解析到$streamData后加json_decode处理，再拆分字段写入MySQL对应列
• 如果SSE接口需要鉴权，可以在cURL配置中添加CURLOPT_HTTPHEADER参数传入请求头，比如token、cookie等
• 脚本中的数据库表结构请根据你实际的业务需求调整，上述代码仅做示例

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