Lumen开发：结合Redis实现消息队列（2）

上一篇讲了Lumen配置Redis，现在来讲一下，如何实现消息队列

2、编写任务类

2.1  任务类结构

默认情况下，应用的所有队列任务都存放在app/Jobs目录。任务类非常简单，正常情况下只包含一个当队列处理该任务时被执行的handle方法，让我们看一个任务类的例子：、

<?php

namespace AppJobs;

use AppUser;
use AppJobsJob;
//use IlluminateContractsMailMailer;//略过邮箱操作
use IlluminateQueueSerializesModels;
//use IlluminateQueueInteractsWithQueue;//会引起jobs的冲突
//use IlluminateContractsBusSelfHandling;//已自动实现
use IlluminateContractsQueueShouldQueue;

//class SendReminderEmail extends Job implements SelfHandling, ShouldQueue
class SendReminderEmail extends Job implements ShouldQueue
{
    //use InteractsWithQueue, SerializesModels;
    use SerializesModels;

    protected $user;

    /**
     * 创建一个新的任务实例
     *
     * @param  User  $user
     * @return void
     */
    public function __construct(User $user)
    {
        $this->user = $user;
    }

    /**
     * 执行任务
     *
     * @param  Mailer  $mailer
     * @return void
     */
    public function handle(Mailer $mailer)
    {
        sleep(5);
        //TODO 这里可以往数据库表插入一条记录，可以看出异步效果
    }
}

在本例中，注意我们能够直接将Eloquent模型传递到对列任务的构造函数中。由于该任务使用了SerializesModels trait，Eloquent模型将会在任务被执行是优雅地序列化和反序列化。如果你的队列任务在构造函数中接收Eloquent模型，只有模型的主键会被序列化到队列，当任务真正被执行的时候，队列系统会自动从数据库中获取整个模型实例。这对应用而言是完全透明的，从而避免序列化整个Eloquent模型实例引起的问题。

handle方法在任务被队列处理的时候被调用，注意我们可以在任务的handle方法中对依赖进行类型提示。Lumen服务容器会自动注入这些依赖。

出错

如果任务被处理的时候抛出异常，则该任务将会被自动释放回队列以便再次尝试执行。任务会持续被释放知道尝试次数达到应用允许的最大次数。最大尝试次数通过Artisan任务queue:listen或queue:work上的--tries开关来定义。关于运行队列监听器的更多信息可以在下面看到。

手动释放任务

如果你想要手动释放任务，生成的任务类中自带的InteractsWithQueue trait提供了释放队列任务的release方法，该方法接收一个参数——同一个任务两次运行之间的等待时间：

public function handle(Mailer $mailer){
    if (condition) {
        $this->release(10);
    }
}

检查尝试运行次数

正如上面提到的，如果在任务处理期间发生异常，任务会自动释放回队列中，你可以通过attempts方法来检查该任务已经尝试运行次数：

public function handle(Mailer $mailer){
    if ($this->attempts() > 3) {
        //
    }
}

　　

3、推送任务到队列

默认的Lumen控制器位于app/Http/Controllers/Controller.php并使用了DispatchesJobs trait。该trait提供了一些允许你方便推送任务到队列的方法，例如dispatch方法：

?php

namespace AppHttpControllers;

use AppUser;
use IlluminateHttpRequest;
use AppJobsSendReminderEmail;
use AppHttpControllersController;

class UserController extends Controller{
    /**
     * 发送提醒邮件到指定用户
     *
     * @param  Request  $request
     * @param  int  $id
     * @return Response
     */
    public function sendReminderEmail(Request $request, $id)
    {
        $user = User::findOrFail($id);

        $this->dispatch(new SendReminderEmail($user));
    }
}

为任务指定队列

你还可以指定任务被发送到的队列。

通过推送任务到不同队列，你可以对队列任务进行“分类”，甚至优先考虑分配给多个队列的worker数目。这并不会如队列配置文件中定义的那样将任务推送到不同队列“连接”，而只是在单个连接中发送给特定队列。要指定该队列，使用任务实例上的onQueue方法，该方法有Lumen自带的基类AppJobsJob提供：

<?php

namespace AppHttpControllers;

use AppUser;
use IlluminateHttpRequest;
use AppJobsSendReminderEmail;
use AppHttpControllersController;

class UserController extends Controller{
    /**
     * 发送提醒邮件到指定用户
     *
     * @param  Request  $request
     * @param  int  $id
     * @return Response
     */
    public function sendReminderEmail(Request $request, $id)
    {
        $user = User::findOrFail($id);
        $job = (new SendReminderEmail($user))->onQueue('emails');
        $this->dispatch($job);
    }
}

3.1 延迟任务

有时候你可能想要延迟队列任务的执行。例如，你可能想要将一个注册15分钟后给消费者发送提醒邮件的任务放到队列中，可以通过使用任务类上的delay方法来实现，该方法由IlluminateBusQueueable trait提供：

<?php

namespace AppHttpControllers;

use AppUser;
use IlluminateHttpRequest;
use AppJobsSendReminderEmail;
use AppHttpControllersController;

class UserController extends Controller{
    /**
     * 发送提醒邮件到指定用户
     *
     * @param  Request  $request
     * @param  int  $id
     * @return Response
     */
    public function sendReminderEmail(Request $request, $id)
    {
        $user = User::findOrFail($id);
        $job = (new SendReminderEmail($user))->delay(60);
        $this->dispatch($job);
    }
}

在本例中，我们指定任务在队列中开始执行前延迟60秒。

注意：Amazon SQS服务最大延迟时间是15分钟。

3.2 从请求中分发任务

映射HTTP请求变量到任务中很常见，Lumen提供了一些帮助函数让这种实现变得简单，而不用每次请求时手动执行映射。让我么看一下 DispatchesJobs trait上的dispatchFrom方法。默认情况下，该trait包含在Lumen控制器基类中：

<?php

namespace AppHttpControllers;

use IlluminateHttpRequest;
use AppHttpControllersController;

class CommerceController extends Controller{
    /**
     * 处理指定订单
     *
     * @param  Request  $request
     * @param  int  $id
     * @return Response
     */
    public function processOrder(Request $request, $id)
    {
        // 处理请求...
        $this->dispatchFrom('AppJobsProcessOrder', $request);
    }
}

　　

该方法检查给定任务类的构造函数并从HTTP请求（或者其它ArrayAccess对象）中解析变量来填充任务需要的构造函数参数。所以，如果我们的任务类在构造函数中接收一个productId变量，该任务将会尝试从HTTP请求中获取productId参数。

你还可以传递一个数组作为dispatchFrom方法的第三个参数。该数组用于填充所有请求中不存在的构造函数参数：

$this->dispatchFrom('AppJobsProcessOrder', $request, [
    'taxPercentage' => 20,
]);

　　

4、运行队列监听器

开启任务监听器

Lumen包含了一个Artisan命令用来运行推送到队列的新任务。你可以使用queue:listen命令运行监听器：

php artisan queue:listen

　　

每次执行控制器对应方法，执行任务写入，然后再异步执行对应任务，后面两次前后都相隔了5秒，因为加了sleep(5)，不过执行sleep前就返回了结果！

下一篇再详细分析具体的操作方法和处理方法