多Worker多服务器环境下，如何确保Laravel链式任务在同一服务器执行？

针对你遇到的任务链依赖本地文件、必须在同一服务器执行的问题，以下是几个成熟的落地方案：

1. 服务器专属队列分组 #

这是最直接可靠的方案，核心思路是给每台服务器分配专属队列，让任务链的所有任务都投递到该队列，同时服务器的Worker只监听自己的专属队列。

实现步骤： #

• 给每台服务器的.env文件配置唯一标识：

  SERVER_ID=worker-server-01
  

• 任务链的第一个任务（Job1）执行完成后，将后续任务投递到当前服务器的专属队列：

  // Job1的handle方法中
  public function handle()
  {
      // 执行Job1的文件操作逻辑...
  
      // 将Job2投递到当前服务器的专属队列
      dispatch(new Job2())->onQueue(env('SERVER_ID'));
      // 同理处理Job3、Job4、Job5
  }
  

• 启动服务器的Worker时，指定只监听自己的专属队列：

  php artisan queue:work --queue=$(cat .env | grep SERVER_ID | cut -d '=' -f2)
  

  或者直接写死队列名：

  php artisan queue:work --queue=worker-server-01
  

这种方案完全避免了跨服务器调度的问题，任务链的所有任务只会被对应服务器的Worker处理。

2. Redis标记+队列中间件拦截 #

如果不想为每台服务器创建专属队列，可以用Redis记录任务链的归属服务器，再通过队列中间件拦截非归属服务器的Worker处理请求。

实现步骤： #

• 给任务链生成唯一ID，在Job1启动时将链ID与当前服务器ID绑定到Redis：

  // Job1的构造方法或handle方法中
  public function __construct()
  {
      $this->chainId = uniqid('chain_', true);
      // 将链ID与当前服务器ID存入Redis，设置过期时间（根据任务链最长执行时间调整）
      Redis::setex("job_chain:{$this->chainId}", 3600, env('SERVER_ID'));
  }
  
  public function handle()
  {
      // 执行Job1逻辑...
  
      // 分发后续任务时带上chainId
      dispatch(new Job2($this->chainId))->onQueue('default');
  }
  

• 创建队列中间件StickyJobMiddleware，用于拦截非归属服务器的任务：

  <?php
  
  namespace App\Http\Middleware;
  
  use Closure;
  use Illuminate\Support\Facades\Redis;
  
  class StickyJobMiddleware
  {
      public function handle($job, Closure $next)
      {
          // 从任务中获取chainId
          $chainId = $job->chainId ?? null;
          if (!$chainId) {
              return $next($job);
          }
  
          $assignedServer = Redis::get("job_chain:{$chainId}");
          // 如果任务不属于当前服务器，将任务重新放回队列，跳过处理
          if ($assignedServer !== env('SERVER_ID')) {
              $job->release(5); // 5秒后重试，避免频繁争抢
              return;
          }
  
          // 属于当前服务器，执行任务
          $next($job);
  
          // 任务链最后一个任务执行完成后，删除Redis中的标记
          if ($job instanceof Job5) {
              Redis::del("job_chain:{$chainId}");
          }
      }
  }
  

• 在所有任务链的Job类中注册该中间件：

  use App\Http\Middleware\StickyJobMiddleware;
  
  class Job2 implements ShouldQueue
  {
      public $middleware = [StickyJobMiddleware::class];
      public $chainId;
  
      public function __construct($chainId)
      {
          $this->chainId = $chainId;
      }
  
      // ... handle方法
  }
  

• 每台服务器的Worker正常启动即可，中间件会自动过滤不属于自己的任务。

3. 同步执行任务链（适合短耗时任务） #

如果任务链的总执行时间较短（不会阻塞Worker或请求），可以在Job1中直接同步执行后续任务，完全避免异步队列的分布式调度问题：

// Job1的handle方法中
public function handle()
{
    // 执行Job1逻辑...

    // 同步执行Job2到Job5
    (new Job2())->handle();
    (new Job3())->handle();
    (new Job4())->handle();
    (new Job5())->handle();
}

这种方案最简单，但要注意如果任务链耗时过长，会导致Worker被长时间占用，影响其他任务处理。

补充：为什么你之前的Machine ID方案无效？ #

你之前在任务负载中附加Machine ID，但没有限制Worker的拾取规则，分布式队列的Worker仍然会随机获取任务，所以无法保证归属。必须配合任务投递到指定队列或Worker拦截非归属任务的逻辑，才能实现任务链的服务器绑定。

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