如何实现Redis+Horizon队列任务串行执行及分页任务追加？

兄弟，你这个需求我太懂了——要处理有依赖顺序的分片数据，还得支持后续分页数据动态追加到任务链，之前重复三次入队的方式确实有点糙，而且没法保证绝对的顺序（万一某个任务执行慢了，后面延迟到的任务就会抢跑）。我给你分享几个更优雅的方案，绝对适配你的Redis+Horizon环境：

核心思路：用「任务完成触发下一个」替代固定延迟 #

你的核心痛点是必须严格按顺序执行分片，而且要支持后续分页数据无缝追加。最靠谱的方式是让每个任务处理完后，主动触发下一个任务，而不是提前设置一堆延迟。结合Redis来暂存待执行的分片，完美解决动态追加的问题。

步骤1：改造ImportData任务，实现链式触发 #

修改任务的handle方法，处理完当前分片后，从Redis的待执行队列里取下一个分片并调度新任务：

<?php

namespace App\Jobs;

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;

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

    protected $dataPart;

    public function __construct($dataPart)
    {
        $this->dataPart = $dataPart;
    }

    // 可根据需求设置重试次数，避免单个分片失败中断整个链
    public $tries = 3;

    public function handle()
    {
        // 👇 这里写你的逐行处理逻辑，保证当前分片的行依赖都满足
        foreach ($this->dataPart as $row) {
            // 处理单条数据...
        }

        // 处理完后，尝试获取下一个分片并触发任务
        $nextPartJson = Redis::lpop('import_data_pending');
        if ($nextPartJson) {
            $nextDataPart = json_decode($nextPartJson, true);
            // 调度下一个任务，继续链式执行
            ImportData::dispatch($nextDataPart);
        } else {
            // 所有分片处理完成，可以做一些收尾操作（比如更新状态通知客户端）
            Redis::del('import_data_running');
        }
    }
}

步骤2：处理API请求，将分片存入Redis并启动任务链 #

不管是第一次请求还是后续分页请求，都把分片追加到Redis的待执行队列，然后检查是否需要启动第一个任务（避免重复启动）：

// 你的API控制器方法
public function import(Request $request)
{
    $rawData = $request->input('data');
    // 拆分成250行的分片
    $dataParts = array_chunk($rawData, 250);

    // 将分片序列化后追加到Redis队列（用rpush保证先提交的分片先执行）
    foreach ($dataParts as $part) {
        Redis::rpush('import_data_pending', json_encode($part));
    }

    // 用Redis锁防止多实例/多请求重复启动第一个任务
    $lockKey = 'import_data_running';
    if (Redis::setnx($lockKey, true)) {
        // 设置锁过期时间，防止任务失败后锁永久存在
        Redis::expire($lockKey, 3600);
        
        // 取出第一个分片启动任务链
        $firstPartJson = Redis::lpop('import_data_pending');
        if ($firstPartJson) {
            $firstDataPart = json_decode($firstPartJson, true);
            // 用afterCommit确保事务提交后再触发任务（如果你的请求有数据库事务的话）
            ImportData::dispatch($firstDataPart)->afterCommit();
        } else {
            // 没有分片可处理，释放锁
            Redis::del($lockKey);
        }
    }

    return response()->json(['status' => 'accepted', 'message' => '数据已加入处理队列']);
}

这个方案的优势 #

• 绝对顺序执行：只有前一个任务处理完成，才会触发下一个，完全满足你「每行依赖前面行」的要求，不会出现抢跑问题。
• 支持动态追加：后续分页请求的分片直接追加到Redis队列末尾，前面的任务执行完会自动处理新分片，不用关心当前执行进度。
• 资源利用率更高：不用像之前那样提前重复入队三次，而是按需调度任务，避免无效的重复执行。
• Horizon友好：所有ImportData任务都能在Horizon里正常监控，执行状态、重试记录一目了然。

额外优化建议 #

• 如果你的分片处理逻辑可能出现异常，可以给任务加上failed方法，处理失败时的回滚或通知：

  public function failed(\Throwable $exception)
  {
      // 记录日志、通知管理员，或者把失败的分片重新放回队列
      Redis::rpush('import_data_pending', json_encode($this->dataPart));
  }
  

• 可以给Redis的待执行队列加上前缀（比如import_data_pending:{$user_id}），区分不同用户的导入任务，避免互相干扰。

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