PHP - Yii2 异步队列

1. 前言使用场景

在 PHP Yii2 中，队列是一种特殊的数据结构，用于处理和管理后台任务。队列允许我们将耗时的任务（如发送电子邮件、push通知等）放入队列中，然后在后台异步执行。这样可以避免在处理大量请求时阻塞主应用程序，从而提高应用程序的性能和响应能力。

批量任务需要提高处理速度，比如10万个消息，可以起10个异步队列消费进程，同时消费。

延时任务，如订单超时30分钟未支付，自动取消。

在 Yii2 中，队列的实现通常使用 Supervisor 或 Guzzle 这样的守护进程来监听队列，并在有新任务时自动执行。队列中的任务通常以闭包函数或类的实例形式存在，可以指定任务的处理顺序、优先级等。

使用 Yii2 队列时，你需要进行以下步骤：

1. 配置队列组件：在应用程序的配置文件中，你需要配置队列组件的连接信息，包括队列服务器地址、端口、驱动方式等。
2. 创建任务：创建一个继承自 yii\queue\JobInterface 的类，实现 execute() 方法来定义具体的任务逻辑。可以将任务中的耗时操作放在这个方法中执行。
3. 推送任务：使用 Yii::$app->queue->push() 方法将任务推送到队列中。你可以根据需要指定任务的优先级、延迟执行时间等选项。
4. 启动队列监听：通过运行 php yii queue/listen 命令来启动队列监听。这个命令会监听队列中的新任务，并在有新任务时自动执行。你可以根据需要指定监听的进程数、超时时间等参数。
5. 处理队列任务：当有新任务到达时，监听进程会调用任务类中的 execute() 方法来执行任务。任务完成后，监听进程会自动处理下一个任务。

通过使用 Yii2 队列，你可以将耗时的后台任务与主应用程序分离，提高应用程序的性能和响应能力。同时，你还可以通过监听队列来控制任务的执行顺序、优先级等，实现更加灵活的任务调度。

2. 使用方法代码片段

1. 消费端

1. 在配置文件中添加队列配置，console/config/main.php

//配置
'bootstrap' => [            'log',             'queueConf', // 把这个组件注册到控制台        
],        
'redisConf' => [            'class' => 'yii\redis\Connection',            'hostname' => '127.0.0.1',            'port' => 6379,            'database' => 0        
],        
'queueConf' => [            'class' => yii\queue\redis\Queue::class,            'redis' => 'redisConf', // 队列使用的redis            'as log' => \yii\queue\LogBehavior::class, //日志            'channel' => 'queueConf',        
],

   2. 添加配置之后可以运行./yii myqueue/info，可以看到队列当前状态

   3. 启动./yii myqueue/listen 开始监听，可启动多个进程同时消费

2. 生产端

1.在写队列的模块下，也添加配置，这里不需要加 bootstrap

'conf_redis' => [            'class' => 'yii\redis\Connection',            'hostname' => '127.0.0.1',            'port' => 6379,            'database' => 0        
],        
'queue_conf' => [            'class' => yii\queue\redis\Queue::class,            'redis' => 'conf_redis', // 队列使用的redis            'as log' => \yii\queue\LogBehavior::class,            'channel' => 'queue_conf',        
],

2.在common/jobs下，新建一个Job文件，在execute方法中添加要处理的业务逻辑

<?phpnamespace common\jobs;use yii\base\BaseObject;
use yii\queue\JobInterface;class TestJob extends BaseObject implements JobInterface{    public $orderId;    /**     * @param Queue $queue which pushed and is handling the job     */    public function execute($queue)    {        // 业务逻辑        $time = time();        $msg =  "时间：{$time},队列-订单号：{$this->orderId}\n";file_put_contents('/tmp/testjob.log',$msg, FILE_APPEND);    }
}

3.写入队列

<?phpfor($i=0; $i < 10; $i++) {            Yii::$app->queue_conf->push(new TestJob(['orderId' => $timestamp . $i,            ]));        }

4.延时执行

<?php// 5秒后执行
Yii::$app->queue_conf->delay(5)->push(new TestJob([                'orderId' => $timestamp . $i,            
]));

3. 调试方法

配置sync，任务可以同步执行，便于调试

<?php'myqueue' => [            'class' => \yii\queue\sync\Queue::class,            'handle' => true, // 任务是否立即执行        
],

4. 更多参考

    Yii2官方手册