Laravel中如何防止AdjustWalletBalance任务并发执行？

我正在开发一款应用的钱包充值系统，使用AdjustWalletBalance Job处理用户余额调整，逻辑是读取用户当前钱包余额后增加指定金额。由于钱包操作涉及资金安全，为避免同一用户的Job并发执行引发竞态条件、导致余额计算错误，我添加了以用户ID为唯一键的WithoutOverlapping中间件，但该中间件并未生效——同时运行两个队列Worker时，同一用户的AdjustWalletBalance任务仍会并行执行，最终余额计算错误。

路由代码示例 #

Route::get('/queue-test', function() {
    dispatch(new \App\Jobs\Tenant\AdjustWalletBalance(50, User::first()));
});

错误日志信息 #

[2023-09-24 19:12:42] local.DEBUG: starting 1007152283 at: 2023-09-24 19:12:42  
[2023-09-24 19:12:42] local.DEBUG: adjusting wallet balance with amount: 50  
[2023-09-24 19:12:42] local.DEBUG: current balance for 1007152283: 0  
[2023-09-24 19:12:43] local.DEBUG: starting 241490440 at: 2023-09-24 19:12:43
[2023-09-24 19:12:43] local.DEBUG: adjusting wallet balance with amount: 50  
[2023-09-24 19:12:43] local.DEBUG: current balance for 241490440: 0  
[2023-09-24 19:12:52] local.DEBUG: 1007152283 updated balance to 50  
[2023-09-24 19:12:53] local.DEBUG: 241490440 updated balance to 50  

已尝试的无效操作 #

• 将CACHE_DRIVER设置为array
• 实现ShouldBeUnique契约并指定用户ID为唯一标识
• 单Worker运行时任务可正常排队，但应用需支持数百用户，必须运行多Worker，因此必须解决该并发问题

解决方法 #

1. 替换为共享缓存驱动 #

array缓存是进程内缓存，多Worker属于不同进程，缓存无法共享，导致队列锁完全失效。必须改用跨进程共享的缓存驱动，比如Redis、Memcached：

修改.env配置：

CACHE_DRIVER=redis

2. 正确配置WithoutOverlapping中间件 #

确保Job中正确绑定中间件，并且锁的唯一键为用户ID，同时设置锁过期时间防止死锁：

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

    protected $userId;
    protected $amount;

    public function __construct(int $amount, User $user)
    {
        $this->amount = $amount;
        $this->userId = $user->id;
    }

    public function middleware()
    {
        return [
            (new WithoutOverlapping($this->userId))->expireAfter(30)
        ];
    }

    public function handle()
    {
        // 数据库原子更新操作，避免先读再写的竞态
        User::where('id', $this->userId)->increment('wallet_balance', $this->amount);
    }
}

3. 数据库层面强制原子更新 #

队列锁是应用层防护，数据库层面必须用原子更新彻底杜绝竞态条件，禁止先读取余额再计算保存的写法。推荐两种原子更新方式：

方式一：使用increment方法（最简洁）

User::where('id', $this->userId)->increment('wallet_balance', $this->amount);

方式二：使用DB::raw自定义更新逻辑

User::where('id', $this->userId)->update([
    'wallet_balance' => DB::raw('wallet_balance + ?', [$this->amount])
]);

4. 正确实现ShouldBeUnique契约（可选） #

如果选择使用ShouldBeUnique替代WithoutOverlapping，需确保唯一标识正确，且缓存驱动为共享类型：

class AdjustWalletBalance implements ShouldQueue, ShouldBeUnique
{
    // ... 其他属性和构造方法

    public function uniqueId()
    {
        return 'adjust-wallet:' . $this->userId;
    }

    public function handle()
    {
        User::where('id', $this->userId)->increment('wallet_balance', $this->amount);
    }
}

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