Laravel使用Bus Batch处理Shopify缓存时pending jobs出现负数如何解决

安全阿岳

2026-9-27

问题原因

核心原因1：你先调用Bus::batch([])->dispatch()派发了空批次，之后再动态往批次里追加任务，Laravel批次的计数逻辑默认是创建时就统计总任务数，动态追加时如果队列消费者已经在处理初始空批次，会出现计数不同步。
核心原因2：批次创建和任务追加逻辑被包裹在数据库事务中，事务提交前批次的更新数据不会持久化，但队列进程可能已经读取到未提交的批次状态，导致计数错乱。
核心原因3：任务handle方法中限流失败后调用$this->release(10)重发任务时，Laravel默认批次计数会直接扣减待处理任务数，重发的任务不会重新计入待处理统计，最终导致待处理数变为负数。
代码bug：控制器中追加任务的匿名函数引入了未定义的$merchant变量，会导致任务创建失败，进一步打乱批次计数。
参数不合理：Redis::throttle的block时间设为70秒，超过了60秒的限流窗口，会导致大量任务长时间阻塞抢锁，触发更多重发逻辑。

解决方案

修改批次创建逻辑
不要先派发空批次，先循环拉取所有Shopify产品，把所有CacheProducts任务实例预先收集到数组中，再一次性创建并派发批次：

// 先收集所有任务
$jobs = [];
$since_id = 0;
while ($since_id >= 0) {
    // 拉取产品逻辑不变
    $fetchedProducts = (new ShopifyProduct(
        $sample->shopify_domain,
        $sample->shopify_info['access_token'])
    )
    ->getById($since_id)
    ->then(function ($data) {
        return $data['products'];
    }, function () {
        return [];
    })
    ->wait();

    if (collect($fetchedProducts)->count() == 0) break;

    $lastProduct = Arr::last($fetchedProducts);
    $since_id = $lastProduct['id'];

    foreach ($fetchedProducts as $shopifyProduct) {
        // 修复未定义$merchant的bug，替换为你实际需要的变量
        $jobs[] = new CacheProducts($shopifyProduct, $sample);
    }
}
// 一次性创建批次
$batch = Bus::batch($jobs)->dispatch();

移除批次逻辑外层的数据库事务
批次相关操作不需要和业务逻辑放在同一个事务里，避免事务未提交导致的状态不同步，仅把需要原子性保证的业务操作包裹在事务中即可。
调整限流逻辑和重发处理
把限流逻辑移到任务的middleware方法中，使用Laravel自带的限流中间件，会自动处理重发任务的批次计数问题：

// 在CacheProducts任务类中添加middleware方法
public function middleware()
{
    return [
        (new \Illuminate\Queue\Middleware\RateLimited('shopify-cache'))
            ->allow(30)
            ->every(60)
            ->backoff(10) // 重发等待时间
    ];
}

// handle方法中移除throttle相关代码，只保留业务逻辑
public function handle()
{
    (new Metafield(
            $this->sample->shopify_domain, // 修复$merchant变量问题
            $this->sample->shopify_info['access_token']
        )
    )
    ->get($this->shopifyProduct['id'], 'products')
    ->then(function ($data) {
        $this->shopifyProduct['metafields'] = $data['metafields'];
        $cacheProducts = Cache::tags($this->sample->name)->get('products') ?? [];
        array_push($cacheProducts, $this->shopifyProduct);
        Cache::tags($this->sample->name)->put('products', $cacheProducts, now()->addHour());
    })
    ->wait();
}

调整限流参数
不要设置超过限流窗口的阻塞时间，限流窗口60秒的情况下，阻塞时间最多设置为30秒即可，避免长时间占用队列进程。

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