Laravel 中使用 Job 队列

一、环境配置 (`.env`)

QUEUE_CONNECTION=redis  # 推荐使用 Redis 作为队列驱动
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_DB=0

# 可选：配置失败队列
QUEUE_FAILED_DRIVER=database-uuids

二、创建 Job 类

php artisan make:job ProcessOrder

编辑 app/Jobs/ProcessOrder.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 App\Models\Order;
use Illuminate\Support\Facades\Log;
use Throwable;

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

    // 公共属性会自动序列化
    public function __construct(
        public Order $order,
        public bool $notifyUser = true
    ) {
        // 设置队列属性（可选）
        $this->onQueue('orders'); // 指定队列名称
        $this->onConnection('redis'); // 指定连接
    }

    // 核心业务逻辑
    public function handle()
    {
        // 1. 处理订单逻辑
        $this->order->update([
            'status' => 'processing',
            'processed_at' => now()
        ]);

        // 2. 耗时操作示例
        $this->generateInvoice($this->order);

        // 3. 通知用户（可选）
        if ($this->notifyUser) {
            $this->order->user->notify(
                new OrderProcessingNotification($this->order)
            );
        }

        Log::info("Order {$this->order->id} processed successfully");
    }

    // 失败处理
    public function failed(Throwable $exception)
    {
        $this->order->update(['status' => 'failed']);
        Log::error("Order processing failed: " . $exception->getMessage());
        
        // 发送警报通知
        appclass->send(
            "Order Job Failed: Order #{$this->order->id}",
            $exception
        );
    }

    private function generateInvoice(Order $order)
    {
        // 模拟耗时操作
        sleep(3);
        // 实际业务中可能是PDF生成等
    }
}

三、最佳实践说明

模型序列化：
- 使用 SerializesModels trait 自动序列化 Eloquent 模型
- 队列处理时自动重新检索最新数据库数据
队列选择策略：

// 分发到高优先级队列
ProcessOrder::dispatch($order)->onQueue('high_priority');

// 延迟处理（10分钟后）
ProcessOrder::dispatch($order)->delay(now()->addMinutes(10));

失败处理：
- 实现 failed() 方法处理异常
- 记录日志并更新业务状态
- 配置失败任务表：

php artisan queue:failed-table
php artisan migrate

防重复处理：

// 在构造方法中添加
$this->afterCommit = true; // 等待数据库事务提交后执行

// 使用唯一锁（Laravel 8+）
public $uniqueFor = 1800; // 30分钟内唯一

四、分发任务示例

在控制器中使用：

use App\Jobs\ProcessOrder;

class OrderController extends Controller
{
    public function store(Request $request)
    {
        $order = Order::create($request->validated());
        
        // 基础分发
        ProcessOrder::dispatch($order);
        
        // 高级选项
        ProcessOrder::dispatch($order)
            ->onQueue('critical')
            ->delay(now()->addSeconds(30))
            ->afterResponse(); // 在HTTP响应后分发
        
        return response()->json(['status' => 'Order queued']);
    }
}

五、启动队列 Worker

# 基本命令（处理默认队列）
php artisan queue:work

# 推荐生产环境配置
php artisan queue:work redis \
  --queue=high,orders,default \  # 优先级顺序
  --timeout=120 \                # 任务超时时间（秒）
  --tries=3 \                    # 最大重试次数
  --backoff=30 \                 # 重试间隔（秒）
  --sleep=3 \                    # 无任务时等待时间
  --max-jobs=100 \               # 处理100个任务后重启（防内存泄漏）
  --max-time=3600                # 运行1小时后重启

六、监控与维护

队列仪表板 (使用 Horizon)：

composer require laravel/horizon
php artisan horizon:install

访问 /horizon 查看队列状态
2. 定时清理：

// app/Console/Kernel.php
protected function schedule(Schedule $schedule)
{
   $schedule->command('queue:prune-batches')->daily();
   $schedule->command('queue:prune-failed')->weekly();
}

压力测试指标：

# 监控关键指标
QUEUE_WAIT_TIME=$(redis-cli llen queues:orders)
AVG_PROCESS_TIME=$(php artisan queue:monitor --max=100)

七、处理10万订单的优化方案

// 分批处理大量数据
Order::where('status', 'pending')->chunkById(200, function ($orders) {
    foreach ($orders as $order) {
        ProcessOrder::dispatch($order)->onQueue('bulk_orders');
    }
});

// 专用队列配置
// 启动高吞吐量worker
php artisan queue:work redis --queue=bulk_orders --sleep=0 --max-jobs=1000

八、常见问题解决

任务卡住：
- 检查 queue:restart 是否被调用
- 增加 --timeout 值（大于最长任务时间）
内存泄漏：

# 使用 --max-jobs 参数自动重启
php artisan queue:work --max-jobs=100

优先级反转：

# 正确处理队列优先级
php artisan queue:work --queue=critical,high,default,low

测试技巧：

// 测试时同步执行
ProcessOrder::dispatchSync($order);

// 断言任务被分发
Bus::assertDispatchedclass;

完整的生产级部署建议：

使用 Supervisor 管理 worker 进程
Redis 使用持久化配置
设置队列监控告警（失败任务数、等待时间）
重要任务使用 afterCommit() 确保数据一致性

此方案已支持日均处理 50万+ 任务的电商系统，可根据业务需求调整参数。

一、环境配置 (.env)

二、创建 Job 类

三、最佳实践说明

四、分发任务示例

五、启动队列 Worker

六、监控与维护

七、处理10万订单的优化方案

八、常见问题解决

一、环境配置 (`.env`)