Laravel队列中合并FetchKeywordStatistics作业以减少第三方API请求是否可行？

完全可行，通过临时存储收集关键词+聚合作业批量处理的方式，就能实现合并多个客户端请求的关键词，大幅减少对第三方API的调用次数，同时适配对方的限流和批量限制。

核心思路 #

• 不再直接分发单个关键词处理作业，而是将所有待查询的关键词存入临时存储（如Redis）
• 调度一个聚合作业，延迟一段时间执行，用来收集这段时间内的所有关键词
• 聚合作业执行时，取出所有关键词，去重、分批后调用第三方API，再处理结果返回给客户端

具体实现步骤 #

1. 修改API控制器逻辑：收集关键词并调度聚合作业 #

控制器不再直接分发原有的FetchKeywordStatistics作业，而是将关键词存入Redis集合（自动去重），并确保只有一个聚合作业在调度中：

use Illuminate\Support\Facades\Redis;
use App\Jobs\AggregateKeywordStatistics;

public function fetchStats(Request $request)
{
    // 获取客户端传入的关键词数组
    $keywords = $request->validate([
        'keywords' => 'required|array|min:1'
    ])['keywords'];

    // 将关键词存入Redis集合（自动去重）
    Redis::sadd('pending_keywords', ...$keywords);

    // 加锁避免重复调度聚合作业
    if (!Redis::exists('aggregate_job_lock')) {
        // 锁定1分钟，防止重复触发
        Redis::setex('aggregate_job_lock', 60, 'active');
        // 延迟5秒执行，给时间收集更多请求
        AggregateKeywordStatistics::dispatch()->delay(now()->addSeconds(5));
    }

    // 返回查询标识，供客户端后续获取结果
    return response()->json([
        'query_id' => uniqid(),
        'message' => '请求已受理，稍后可通过query_id查询结果'
    ]);
}

2. 创建聚合作业：批量处理关键词 #

新建AggregateKeywordStatistics作业，负责取出所有待处理关键词，分批调用第三方API，并处理结果：

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Redis;
use Illuminate\Support\Facades\Http;

class AggregateKeywordStatistics implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public function handle()
    {
        try {
            // 取出所有待处理关键词并清空集合
            $keywords = Redis::smembers('pending_keywords');
            Redis::del('pending_keywords');

            if (empty($keywords)) {
                return;
            }

            // 按第三方限制的100个关键词分批
            $keywordBatches = array_chunk($keywords, 100);

            foreach ($keywordBatches as $batch) {
                // 遵守第三方限流：每分钟最多3次请求
                $this->enforceThirdPartyRateLimit();

                // 调用第三方API获取统计数据
                $response = Http::get('https://third-party-api.com/stats', [
                    'keywords' => implode(',', $batch)
                ]);
                $response->throw(); // 处理请求失败的情况
                $statsData = $response->json();

                // 将结果存入缓存，供客户端查询
                foreach ($batch as $keyword) {
                    Redis::setex(
                        "keyword_stats:{$keyword}",
                        3600, // 缓存1小时
                        json_encode($statsData[$keyword] ?? [])
                    );
                }
            }
        } catch (\Exception $e) {
            // 请求失败时，将关键词放回集合，重新调度作业
            if (!empty($keywords)) {
                Redis::sadd('pending_keywords', ...$keywords);
            }
            // 延迟10秒重试
            self::dispatch()->delay(now()->addSeconds(10));
            throw $e; // 让队列记录错误日志
        } finally {
            // 释放锁，允许下次调度聚合作业
            Redis::del('aggregate_job_lock');
        }
    }

    /**
     * 实现第三方API的限流控制
     */
    protected function enforceThirdPartyRateLimit()
    {
        $throttleKey = 'third_party_api_requests';
        $requestCount = Redis::incr($throttleKey);

        // 第一次请求时设置过期时间（1分钟）
        if ($requestCount === 1) {
            Redis::expire($throttleKey, 60);
        }

        // 超过每分钟3次限制时，等待到下一分钟
        if ($requestCount > 3) {
            $remainingSeconds = Redis::ttl($throttleKey);
            sleep($remainingSeconds + 1);
            // 重置计数
            Redis::set($throttleKey, 1);
            Redis::expire($throttleKey, 60);
        }
    }
}

3. 新增结果查询接口 #

客户端通过之前拿到的query_id（或直接按关键词）查询统计结果：

public function getStatsResult(Request $request)
{
    $keyword = $request->input('keyword');
    $stats = Redis::get("keyword_stats:{$keyword}");

    if (!$stats) {
        return response()->json([
            'status' => 'pending',
            'message' => '统计结果尚未生成，请稍后重试'
        ], 202);
    }

    return response()->json([
        'status' => 'completed',
        'keyword' => $keyword,
        'stats' => json_decode($stats, true)
    ]);
}

关键注意事项 #

• 关键词去重：用Redis集合存储关键词，自动避免重复查询同一个关键词，减少无效请求
• 延迟调度：延迟几秒执行聚合作业，能最大化合并更多客户端的请求，进一步减少第三方API调用次数
• 锁机制：防止多个聚合作业同时执行，避免重复请求第三方API
• 错误重试：请求第三方API失败时，将关键词放回集合并重试，确保数据不丢失
• 限流控制：严格遵守第三方的每分钟3次请求限制，避免被封禁

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