• 定义job interface，代码参考

<?php
/**
 * JobInterface.php class file.
 * @author karnc
 * @date 2020/12/31
 */

namespace common\drivers\interfaces;

interface JobInterface
{

    /**
     * 任务处理方法
     * @return mixed
     */
    public function handle();
}

• 定义redis queue 驱动，代码参考

<?php

namespace common\drivers\queue;
/**
 * RedisQueue.php class file.
 * @author karnc
 */

use common\drivers\interfaces\JobInterface;
use Yii;

class RedisQueue
{
    /**
     * redis服务
     * @var [type]
     */
    protected $redis;

    /**
     * 队列名称
     * @var [type]
     */
    protected $queue_name;


    /**
     * 初始化
     * RedisQueue constructor.
     * @param string $queue_name
     */
    function __construct($queue_name = '')
    {
        $this->redis = Yii::$app->redis;
        $this->queue_name = $queue_name;
    }


    /**
     * 入队
     * @param JobInterface $job
     * @return mixed
     */
    public function push(JobInterface $job)
    {
        return $this->redis->lpush($this->getRedisKey(), serialize($job));
    }

    /**
     * 出队列
     * @return
     */
    public function pull()
    {
        $job_seria = $this->redis->rpop($this->getRedisKey());
        $job       = unserialize($job_seria);

        if (false === $job ||
            ! $job instanceof JobInterface
        ) {
            return false;
        }

        return $job;
    }

    /**
     * 获取队列任务总数
     * @return [type] [description]
     */
    public function count()
    {
        return $this->redis->llen($this->getRedisKey());
    }

    /**
     * 返回redis存储的key名
     * @return [type] [description]
     */
    protected function getRedisKey()
    {
        $name = ! empty($this->queue_name) ? $this->queue_name : $this->getRedisDefaultKeyName();

        return 'queues:' . $name;
    }

    /**
     * 获取默认的key名
     * @return [type] [description]
     */
    protected function getRedisDefaultKeyName()
    {
        return 'default';
    }
}

• 定义redis驱动实例化管理

<?php
/**
 * Manager.php class file.
 * @author karnc
 */

namespace common\drivers\queue;

use Yii;

class Manager
{
    /**
     * 单例模式
     * @var null
     */
    protected static $_static = null;

    /**
     * 工厂模式
     * @param string $queue_name 队列名称
     * @return RedisQueue
     */
    public static function factory($queue_name = null)
    {

        $queue = new RedisQueue($queue_name);
        return clone $queue;
    }

    /**
     * 单例模式
     * @param  [type] $redis      [description]
     * @param  [type] $queue_name [description]
     * @return [type]             [description]
     */
    public static function instance($queue_name = null)
    {
        if (self::$_static == null) {
            self::$_static = self::factory($queue_name);
        }

        return self::$_static;
    }
}

• 一个简单的job参考

<?php
/**
 * SimpleJob class file.
 * @author karnc
 */

namespace common\jobs;

class SimpleJob implements \common\drivers\interfaces\JobInterface
{

    public $args = [];

    public function __construct(array $args = [])
    {
        $this->args = $args;
    }

    /**
     * 处理队列
     * @return mixed|void
     */
    public function handle(): void
    {
        print_r($this->args);
        echo "Hello World\n";
    }
}

• job助手类封装参考

<?php
/**
 * JobHelper.php class file.
 * @author karnc
 */

use common\drivers\queue\Manager;
use common\jobs\SimpleJob;
use Yii;

class JobHelper
{
    
    /**
     * 投递到队列
     * @param array $args
     */
    public static function toSimple(array $args)
    {
        $spider = new SimpleJob($args);
        Manager::factory(Yii::$app->params['queue']['simple'])->push($spider);
    }
    
}

• 队列消费与守护进行相关代码

Daemon.php

<?php
/**
 * Daemon.php class file.
 * @author karnc
 */
namespace console\components\daemons\simple;

use Yii;

class Daemon
{
    /**
     * 启动守护进程
     * @param bool $daemon
     * @return bool
     */
    public static function start(bool $daemon = false): bool
    {
        $processIdPath = Yii::getAlias(Yii::$app->params['daemon']['simple']['masterPidFilePath']);

        if (is_file($processIdPath))
        {
            /**
             * 进程id存在，则先杀尝试杀掉log中记录的进程id
             */
            $processArr = file($processIdPath);
            foreach($processArr as $processId)
            {
                $processId = trim($processId);
                ($processId > 0) && @posix_kill($processId, SIGTERM);
            }
        }
        $masterProcessor = new Master($daemon);
        $masterProcessor->run();
        return TRUE;
    }

    /**
     * 关闭守护进程
     */
    public static function stop(): void
    {
        $processIdPath = Yii::getAlias(Yii::$app->params['daemon']['simple']['masterPidFilePath']);

        if (is_file($processIdPath))
        {
            $processArr = file($processIdPath);

            foreach($processArr as $processId)
            {
                $processId = trim($processId);
                ($processId > 0) && @posix_kill($processId, SIGTERM);
            }

            @unlink($processIdPath);
        }
    }
}

Master.php

<?php
/**
 * Master.php class file.
 * @author karnc
 */

namespace console\components\daemons\simple;
use Yii;
use swoole\Process;

class Master
{
    /**
     * 子进程列表
     * @var array
     */
    protected $processList = [];

    /**
     * 主循环定时器
     * @var integer
     */
    protected $timerId;

    protected $daemon = false;


    protected $processIdPath;

    /**
     * 构造方法
     * @param bool $daemon
     */
    public function __construct(bool $daemon =false)
    {

        $this->daemon = $daemon;
        $this->processIdPath = Yii::getAlias(Yii::$app->params['daemon']['simple']['masterPidFilePath']);
        if($this->daemon == true) Process::daemon();
        swoole_set_process_name(Yii::$app->params['daemon']['simple']['masterName']);
        file_put_contents($this->processIdPath, posix_getpid().PHP_EOL);
    }

    /**
     * 执行
     */
    public function run(): void
    {

        $this->createWorker(Yii::$app->params['daemon']['simple']['workerNumber']);

//        Process::signal(SIGCHLD, [$this, 'handleForSIGCHLD']);
//        Process::signal(SIGTERM, [$this, 'handleForSIGTERM']);

        $this->timerId = swoole_timer_tick(6000, function($timerId, $params = NULL) {

        });

    }

    /**
     * 生成WORKER进程
     * @param int $number
     * @param bool $daemon
     */
    protected function createWorker(int $number): void
    {
        while ($number > 0)
        {
            $number--;

            $process = new Process(
                function(Process $process) use ($number) {
                    $worker = new Worker();
                    $worker->run();
                },
                false,
                0 //子进程内没有进程间通信需求
            );
            if($this->daemon == true) Process::daemon();

            $processId = $process->start();
            file_put_contents($this->processIdPath, $processId.PHP_EOL,FILE_APPEND);
            $this->processList[$processId] = $processId;
        }
    }

    /**
     * 子进程退出时信号响应方法
     * @param int $singalNumber
     */
    public function handleForSIGCHLD(int $singalNumber): void
    {
        while ($result = Process::wait(FALSE))
        {
            unset($this->processList[$result['pid']]);

            $this->createWorker(1);
        }
    }

    /**
     * 主进程退出时信号响应方法
     * @param int $singalNumber
     * @throws \yii\base\ExitException
     */
    public function handleForSIGTERM(int $singalNumber): void
    {
        $this->killChildren();

        swoole_timer_clear($this->timerId);

        while ($result = Process::wait(TRUE))
        {
            // do nothing
        }

        Yii::$app->end();
    }

    /**
     * 杀死子进程
     */
    protected function killChildren(): void
    {
        foreach ($this->processList as $processId)
        {
            Process::kill($processId, SIGTERM);
        }
    }
}

Worker.php

<?php
/**
 * Worker.php class file.
 * @author karnc
 */

namespace console\components\daemons\simple;

use common\drivers\queue\Manager;
use yii\helpers\FileHelper;
use Yii;

pcntl_async_signals(TRUE);

class Worker
{

    /**
     * 构造方法
     */
    public function __construct()
    {
        /**
         * 设置work进程名称
         */
        swoole_set_process_name(Yii::$app->params['daemon']['simple']['workerName']);
    }


    /**
     * work进程执行
     * @throws \yii\base\ExitException
     * @throws \yii\base\InvalidConfigException
     */
    public function run(): void
    {
        while (true)
        {
            $job = $this->process();
            //如没有任务，则休息2秒
            if(!$job)
            {
                sleep(2);
            }
        }
    }


    /**
     * 处理
     * @param array $args
     * @return bool|mixed
     */
    private function process(array $args=[])
    {
        /**
         * 设置阻塞信号
         */
//        pcntl_sigprocmask(SIG_BLOCK, [SIGINT, SIGQUIT, SIGTERM]);

        /**
         * 从redis队列中取任务
         */
        Yii::$app->redis->setex('super_simple_work_time', 86400, time());
        $this->writeSpiderLog('进程成功得到执行');
        $job = Manager::instance(Yii::$app->params['queue']['simple'])->pull();
        $this->writeSpiderLog('成功消费到redis数据');
        if($job) {
            try {
                call_user_func_array([$job, 'handle'], []);
            } catch (\Exception $e) {
                echo $e->getMessage();
            }
        }

        /**
         * 多个进程同时执行消费队列
         */
//        pcntl_sigprocmask(SIG_UNBLOCK, [SIGINT, SIGQUIT, SIGTERM]);

        return $job;
    }

    /**
     * 文件日志记录
     * @param $msg
     * @throws ServicesException
     */
    public function writeSpiderLog($msg)
    {

        $path = Yii::getAlias('@runtime/api/simple/');

        //目录如果不存就先创建目录
        if(!is_dir($path)) {
            FileHelper::createDirectory($path);
        }

        $cp = fopen($path.'log-'.date('Ymd H:i').'.log', "a+");
        $txt = "【".date('Y-m-d H:i:s',time())."】---".$msg."\r\n";
        fwrite($cp, $txt);
        fclose($cp);

    }
}

• Yii2控制台消息队列消费相关代码 CloudController.php

<?php
/**
 * CloudController.php class file.
 * @author karnc
 */

namespace console\controllers;

use console\components\daemons\simple\Daemon;
use console\components\daemons\simple\Worker;
use yii\console\Controller;
use yii\console\ExitCode;

/**
 * 消费队列
 * Class CloudController
 * @package console\controllers
 *
 */
class CloudController  extends Controller
{
    /**
     * 是否为守护进程模式运行
     * @var bool
     */
    public $daemon = false;

    /**
     * 启动服务
     * @return int
     */
    public function actionStart()
    {
        $worker = (new Worker())->run();
        return ExitCode::UNSPECIFIED_ERROR;
    }


    public function options($actionID)
    {
        return array_merge(parent::options($actionID), ['daemon']);
    }


    /**
     * 关闭服务
     * @return int
     */
    public function actionStop()
    {
        Daemon::stop();
        return ExitCode::OK;
    }
}

公共配置参数参考

<?php
return [
    /**
     * 守护进程
     */
    'daemon' => [
        //simple进程配置
        'simple' => [
            'masterPidFilePath' => '@runtime/pid/simple.process.pid',
            'masterName' => 'simple', //进程名称
            'workerNumber' => 10, //进程数量
            'workerName' => 'simple_work' //work进程名称
        ],
    ],
    'queue' => [
        'simple' => 'jobs:simple', //队列名称
    ],
];

以上就实现了在Yii框架中实现消息队列的投递与消费相关功能。