yii2+rabbitmq实现队列(windows)

1.安装Erlang 官方网站，https://erlang.org/download/otp_versions_tree.html，选择和rabbitmq适配的

2.安装RabbitMQ 官方网站，windows：https://www.rabbitmq.com/docs/install-windows

3.安装下载的Erlang和RabbitMQ

碰到的问题：

1. 开启rabbitmq服务时dos窗口一闪而逝，看了服务进程状态是开启状态立马变成了未开启状态，参考了：

https://blog.csdn.net/qq_40808232/article/details/102861014

重新配置了环境变量，

dos进入rabbitmq安装目录的sbin下，

安装插件，命令：rabbitmq-plugins.bat enable rabbitmq_management，

将 C:\Users\Administrator\.erlang.cookie 同步至C:\Windows\System32\config\systemprofile\.erlang.cookie 

同时删除：C:\Users\Administrator\AppData\Roaming\RabbitMQ目录

输入命令：rabbitmq-plugins.bat enable rabbitmq_management ，检查插件安装成功

这时候应该能进rabbitmq管理后台了（http://localhost:15672）

2. 当前php环境没有php_amqp扩展（有的话应该能跳过）

下载地址：https://pecl.php.net/package/amqp/1.11.0/windows，我下的是1.11.0，选择和本地php版本适配的下载，存在非线程安全 (NTS) ；线程安全 (TS)，这个不是很懂，就都下载下来试了一下（试了一下下面的复制转移）

参考了

https://segmentfault.com/q/1010000015206078

复制php_amqp.dll 到php/ext/
复制rabbitmq.4.dll 到php/
复制rabbitmq.4.dll 到windows/system32/ (如32位) ,64位则windows/SysWOW64/

php.ini 加入
extension=php_amqp.dll

重启服务

3. 怎么通过yii2项目使用队列功能

composer require enqueue/amqp-lib（消息队列库）

往yii2项目的web.php的components中（用于代码调用），与web.php同级的console.php中，用于监听（应该是的），都添加同一个队列配置

'test_queue' => [
            'class' => yii\queue\amqp_interop\Queue::class,
            'host' => env('rabbitmq_host'),
            'port' => env('rabbitmq_port'),
            'user' => env('rabbitmq_username'),
            'password' => env('rabbitmq_password'),
            'queueName' => 'test_queue',
            'exchangeName' => 'test_exchange',
            'connectionTimeout' => 5,
            'readTimeout' => 5,
            'serializer' => yii\queue\serializers\JsonSerializer::class,
            'driver' => yii\queue\amqp_interop\Queue::ENQUEUE_AMQP_LIB,
            'as log' => \app\behavior\TaskBehavior::class,
            'heartbeat' => 30,
        ],

5672：client端通信端口（应该是能更改的，没有研究）
15672：管理界面ui端口

上面的rabbitmq_port使用的是client端口5672，其他的rabbitmq参数都换成自己的

继续TaskBehavior文件创建一个：

<?php

namespace app\behavior;

use app\utils\LoggerUtil;
use yii\helpers\ArrayHelper;
use yii\queue\ExecEvent;
use yii\queue\JobEvent;
use yii\queue\JobInterface;
use yii\queue\LogBehavior;
use yii\queue\PushEvent;

/**
 * Class TaskBehavior
 * @package app\behavior
 */
class TaskBehavior extends LogBehavior
{
    /**
     * @param PushEvent $event
     */
    public function afterPush(PushEvent $event)
    {
        $this->log($event, 'push');
    }

    public function beforeExec(ExecEvent $event)
    {
        $this->log($event, 'started');
    }

    /**
     * @param ExecEvent $event
     */
    public function afterExec(ExecEvent $event)
    {
        $this->log($event, 'execute');
    }

    public function afterError(ExecEvent $event)
    {
        $this->log($event, 'execute_error');
    }

    public function getJobParams(JobEvent $event, string $status)
    {
        $job = $event->job;
        $params = ArrayHelper::toArray($job);
        $name = $event->job instanceof JobInterface ? get_class($event->job) : 'unknown job';
        $id = $event->id;

        return [
            'job' => $name,
            'id' => $id,
            'params' => $params,
            'status' => $status,
        ];
    }

    protected function log(JobEvent $event, string $status)
    {
        $data = $this->getJobParams($event, $status);
        $filename= 'debug.log';
        if ($status == 'execute_error') {
            $data['error'] = $event->error;
            $filename = 'logstash.log';
        }

        LoggerUtil::logstash('queue', 'debug', $data, $filename);
    }


}

创建完job文件就能直接调用了

\Yii::$app->test_queue->push(new TestJob([
            'time' => time()
        ]));

testJob文件:

<?php

namespace app\jobs;

use app\utils\LoggerUtil;
use yii\base\BaseObject;
use yii\queue\JobInterface;

class TestJob extends BaseObject implements JobInterface
{
    public $time;

    public function execute($queue)
    {
        LoggerUtil::log('test', '测试job', [
            'time' => $this->time
        ]);
    }
}

最后重要的一步就是监听，切换到项目目录，php yii test-queue/listen，由于只是本地的，所以没有做supervisor保护进程