每秒记录多源JSON变更数据的高效实现方案咨询

安全阿岳

2026-5-19

嘿 Darren，

看起来随着JSON数据源越来越多，你之前用的「每分钟Cron+PHP循环60次」的方案已经顶不住了——毕竟每次启动PHP进程的开销、串行处理所有数据源的效率瓶颈，再加上单个数据源处理慢就会拖慢整个循环的节奏，肯定会越跑越慢。下面给你几个经过实战验证的最优方案，你可以根据自己的技术栈和服务器资源来选：

方案1：用常驻异步进程替代Cron+循环

这是最直接解决“每秒执行”需求的方案，核心是让一个进程一直运行，每秒触发一次检查任务，并且用协程并行处理所有数据源，避免串行阻塞。

如果你熟悉PHP的异步框架，比如Swoole或者Workerman，写个守护进程（Daemon）非常简单：

用框架的定时器功能，每秒触发一次任务；
每个JSON数据源分配一个独立协程去处理拉取、对比、入库逻辑；
用异步HTTP客户端代替file_get_contents，减少网络等待时间。

举个简化的Swoole示例：

use Swoole\Coroutine;
use Swoole\Timer;

// 启动协程环境
Coroutine\run(function () {
    // 每秒执行一次批量检查
    Timer::tick(1000, function () {
        // 假设这是你的数据源列表
        $dataSources = [
            ['id' => 'source_1', 'url' => 'https://example.com/data1.json'],
            ['id' => 'source_2', 'url' => 'https://example.com/data2.json'],
            // ...更多数据源
        ];

        foreach ($dataSources as $source) {
            // 每个数据源用协程并行处理
            Coroutine::create(function () use ($source) {
                // 1. 异步拉取JSON数据
                $client = new Swoole\Http\Client(parse_url($source['url'], PHP_URL_HOST), 443, true);
                $client->get($source['url'], function ($client) use ($source) {
                    $newData = json_decode($client->body, true);
                    $client->close();

                    // 2. 从Redis缓存获取上次的旧数据
                    $redis = new Redis();
                    $redis->connect('127.0.0.1', 6379);
                    $oldData = json_decode($redis->get('data_cache:' . $source['id']), true);

                    // 3. 对比差异（这里可以优化对比逻辑，只检查关键字段）
                    $diff = array_diff_assoc($newData, $oldData ?? []);

                    // 4. 记录差异到数据库（建议批量插入，减少IO）
                    if (!empty($diff)) {
                        $pdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'pass');
                        $stmt = $pdo->prepare("INSERT INTO change_log (source_id, diff_data, created_at) VALUES (?, ?, NOW())");
                        $stmt->execute([$source['id'], json_encode($diff)]);

                        // 更新缓存为最新数据
                        $redis->set('data_cache:' . $source['id'], $client->body);
                    }
                });
            });
        }
    });
});

优点：没有Cron启动进程的额外开销，协程并行处理能最大化利用服务器资源，不会因为单个数据源慢导致整体延迟；
注意点：需要用Supervisord或Systemd管理进程，防止进程意外退出；要做好日志和监控，排查异常数据源。

方案2：消息队列+多消费者架构

如果你的数据源增长非常快，需要水平扩展，那消息队列架构是更灵活的选择。核心思路是把「检查单个数据源」拆成独立任务，每秒由调度进程把所有数据源的任务投递到队列，再由多个消费者进程并行处理。

具体步骤：

用Redis或RabbitMQ做消息队列；
写一个常驻调度进程，每秒把所有数据源的信息（比如ID、URL）投递到队列；
启动多个消费者进程，每个进程从队列中阻塞取出任务，执行拉取、对比、入库逻辑。

简化的消费者示例：

// 消费者进程（可以启动多个）
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

while (true) {
    // 阻塞从队列取任务（避免空轮询浪费资源）
    $task = $redis->blPop('json_check_queue', 0);
    if ($task) {
        $source = json_decode($task[1], true);
        // 这里执行拉取、对比、入库逻辑，和上面的协程示例类似
        // ...
    }
}

优点：高度可扩展，数据源越多，只需要增加消费者进程就行；故障隔离，单个消费者挂了不影响其他任务；
注意点：需要维护消息队列的稳定性，处理消息重复消费的问题（比如给每个任务加唯一ID，入库时做幂等校验）。

方案3：先优化现有逻辑的核心效率

如果暂时不想大改架构，可以先从优化单任务处理速度入手，能缓解一部分压力：

缓存优化：用Redis存储上次拉取的JSON数据，代替从数据库读旧数据，减少数据库IO；
差异对比优化：不要全量对比所有字段，只对比业务关心的关键字段；或者用更高效的对比算法（比如json_diff扩展）；
数据库批量操作：把多个数据源的差异记录攒成一批，用INSERT INTO ... VALUES (...), (...), ...批量插入，减少数据库连接和写入次数；
JSON解析优化：用PHP原生的ext-json扩展（默认已安装），或者用simdjson扩展（比标准json_decode快数倍）；
串行改并行：在现有Cron脚本里，用pcntl_fork创建子进程并行处理多个数据源，避免串行阻塞。

关键通用注意事项

幂等性：确保重复处理同一个数据源不会生成重复的变更记录，可以用「数据源ID+数据哈希值」作为唯一键，入库时做冲突处理；
监控告警：记录每个数据源的处理时间、成功率，当某个数据源连续多次处理失败或超时，及时告警；
资源限制：协程或进程数量不要超过服务器CPU核心数的2倍，避免过多上下文切换导致性能下降。

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