Laravel进阶系列笔记--(七)Laravel 基础数据库存储的消息队列的使用

作者：温新

时间：2021-07-08

hi，我是温新，一名PHPer。

基于数据库驱动的消息队列

第一步：修改驱动为数据库驱动

文件：.env

QUEUE_CONNECTION=database

第二步：创建任务信息表

1）创建迁移文件

// 消息队列任务迁移文件
php artisan queue:table
// 消息队列任务失败迁移文件
php artisan queue:failed-table

2）执行迁移文件

php artisan migrate --path=database/migrations/2021_07_08_145208_create_jobs_table.php
php artisan migrate --path=database/migrations/2021_07_08_145604_create_failed_jobs_table.php

为了便于理解字段的含义，我为这两个迁移文件添加了备注，字段的含义还是有必要了解一下的。

这两个迁移文件后后可直接执行 php artisan migrate

// 2021_07_08_145208_create_jobs_table.php 队列任务表
Schema::create('jobs', function (Blueprint $table) {
    $table->bigIncrements('id');
    $table->string('queue')->index()->comment('队列名');
    $table->longText('payload')->comment('队列处理信息');
    $table->unsignedTinyInteger('attempts')->comment('队列执行次数');
    $table->unsignedInteger('reserved_at')->nullable();
    $table->unsignedInteger('available_at');
    $table->unsignedInteger('created_at');
});

// 2021_07_08_145604_create_failed_jobs_table.php
// 队列执行失败记录表
Schema::create('failed_jobs', function (Blueprint $table) {
    $table->id();
    $table->string('uuid')->unique();
    $table->text('connection')->comment('连接');
    $table->text('queue')->comment('队列');
    $table->longText('payload')->comment('队列任务');
    $table->longText('exception')->comment('异常');
    $table->timestamp('failed_at')->useCurrent()->comment('队列失败时间');
});

第三步：创建任务类

php artisan make:job SendEamilToUser

该命令会在app/Jobs目录下生成SendEamilToUser.php文件。

编辑SendEamilToUser.php文件进行消息队列任务的处理

<?php
namespace App\Jobs;

use App\Models\User;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldBeUnique;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Log;

class SendEamilToUser implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    public $user = null;

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

    public function handle()
    {
        // 此处执行自定义任务
        Log::info('发送邮件给 【' . $this->user->name . '】 用户成功' );
    }
}

第四步：派发任务给队列

1）定义路由

// web.php

Route::get('queue', 'UserController@queue');

2）添加queue方法并编写任务代码

// UserController.php

use App\Jobs\SendEamilToUser;

public function queue()
{
    $users = \App\Models\User::all();
    foreach ($users as $user) {
        // 为每个用户发送消息的任务
        // 方式一
        //dispatch(new SendEamilToUser($user));
		  // 方式二
        SendEamilToUser::dispatch($user);
    }
    return '派发队列任务成功';
}

打开jobs数据表就会看到相关的任务信息了。

3）访问路由添加队列任务

第五步：执行队列任务

--once执行一个队列任务

php artisan queue:work --once
// 执行后返回信息
[2021-07-08 15:41:05][1] Processing: App\Jobs\SendEamilToUser
[2021-07-08 15:41:05][1] Processed:  App\Jobs\SendEamilToUser

练习的环境中建议从执行一个开始。我这里执行--once命令。

queue:work执行所有队列任务

php artisan queue:work

第六步：查看laravel.log文件是否发送邮件成功

// laravel.log
[2021-07-08 15:41:05] local.INFO: 发送邮件给 【Dell Terry】 用户成功

处理失败的队列任务

队列中的任务有可能会失败，执行失败的任务是删除还是重新执行？

任务过期与超时

--timeout设置每个任务超时时间（s）

php artisan queue:work --timeout=60

每个任务60秒没有执行成功，那么90秒后会重新执行该任务。任务重启时间可以通过config/queue.php进行配置，如现在使用的数据库驱动，配置如下

'database' => [
    'driver' => 'database',
    'table' => 'jobs',
    'queue' => 'default'，
    'retry_after' => 90, // 配置此参数，失败任务的重发时间
    'after_commit' => false,
],

失败的任务处理

第一步：修改SendEamilToUser文件模拟任务失败

// app/Jobs/SendEamilToUser.php

public function handle()
{
    // 该函数不存在，因此队列任务执行失败
    testQueue();
    // 此处执行自定义任务
    Log::info('发送邮件给 【' . $this->user->name . '】 用户成功' );
}

任务失败会后进入到failed_jobs数据表中。

第二步：处理失败的任务

// app/Jobs/SendEamilToUser.php
// 只要有队列任务执行失败，就会自动执行该方法
public function failed(\Throwable $exception)
{
    Log::error('发送给 【【' . $this->user->name . '】】的邮件失败');
}

第三步：处理failed_jobs失败的任务（根据需要来执行）

1）queue:failed查看failed_jobs所有失败的任务。该命令将会列出任务 ID、连接、队列和失败时间，任务 ID 可用于重试失败任务。

php artisan queue:failed

2）queue:retry ID重试失败的任务。

php artisan queue:retry d2111845-9d25-4d9a-b00e-67658fe1a9c7

3）queue:retry all重试所有失败的任务

php artisan queue:retry all

4）queue:forget ID删除一个失败的任务

php artisan queue:forget e81f9536-f3e0-49e2-8c38-780d7f242b19

5）queue:flush删除所有失败的任务

php artisan queue:flush

1）这里的ID指的是failed_jobs表中的UUID；

2）我这里执行的是删除所有失败的任务

任务分发

前面演示的是立即分发任务，有时间需要指定时间分发任务，Laravel中提供了相关方法。

延时分发任务

方法：delay(时间)

// app/Http/Controllers/UserController.php

public function queue()
{
    $users = \App\Models\User::all();
    foreach ($users as $user) {
        // 延迟一分钟执行。队列任务最多延时15分钟
        SendEamilToUser::dispatch($user)->delay(now()->addMinutes(1));
    }
    return '派发队列任务成功';
}

同步分发

dispatchNow()同步分发。就是任务不会写入到数据表，而是执行被执行。

public function queue()
{

    $users = \App\Models\User::all();
    foreach ($users as $user) {
        SendEamilToUser::dispatchNow($user);
    }
    return '派发队列任务成功';
}

本篇文章，未完。

我是温新

每天进步一点点，就一点点