Yii2 队列扩展

return [
    'bootstrap' => [
        'queue', // 把这个组件注册到控制台
    ],
    'components' => [
        'queue' => [
            'class' => \yii\queue\<driver>\Queue::class,
            'as log' => \yii\queue\LogBehavior::class,  // 事件日志
            // 驱动的其他选项
        ],
    ],
];

class DownloadJob extends BaseObject implements \yii\queue\JobInterface
{
    public $url;
    public $file;
    
    public function execute($queue)
    {
        file_put_contents($this->file, file_get_contents($this->url));
    }
}

Yii::$app->queue->push(new DownloadJob([
    'url' => 'http://example.com/image.jpg',
    'file' => '/tmp/image.jpg',
]));

Yii::$app->queue->delay(5 * 60)->push(new DownloadJob([
    'url' => 'http://example.com/image.jpg',
    'file' => '/tmp/image.jpg',
]));

public function delay($value)
{
    $this->pushDelay = $value;
    return $this;
}

public function push($job)
{
    $event = new PushEvent([
        'job' => $job,
        'ttr' => $this->pushTtr ?: (
            $job instanceof RetryableJobInterface
                ? $job->getTtr()
                : $this->ttr
        ),
        'delay' => $this->pushDelay ?: 0,
        'priority' => $this->pushPriority,
    ]);
    $this->pushTtr = null;
    $this->pushDelay = null;
    $this->pushPriority = null;

    $this->trigger(self::EVENT_BEFORE_PUSH, $event);
    if ($event->handled) {
        return null;
    }

    if ($this->strictJobType && !($event->job instanceof JobInterface)) {
        throw new InvalidArgumentException('Job must be instance of JobInterface.');
    }

    if (!is_numeric($event->ttr)) {
        throw new InvalidArgumentException('Job TTR must be integer.');
    }
    $event->ttr = (int) $event->ttr;
    if ($event->ttr <= 0) {
        throw new InvalidArgumentException('Job TTR must be greater that zero.');
    }

    if (!is_numeric($event->delay)) {
        throw new InvalidArgumentException('Job delay must be integer.');
    }
    $event->delay = (int) $event->delay;
    if ($event->delay < 0) {
        throw new InvalidArgumentException('Job delay must be positive.');
    }

    $message = $this->serializer->serialize($event->job);
    $event->id = $this->pushMessage($message, $event->ttr, $event->delay, $event->priority);
    $this->trigger(self::EVENT_AFTER_PUSH, $event);

    return $event->id;
}

protected function pushMessage($message, $ttr, $delay, $priority)
{
    $this->db->createCommand()->insert($this->tableName, [
        'channel' => $this->channel,
        'job' => $message,
        'pushed_at' => time(),
        'ttr' => $ttr,
        'delay' => $delay,
        'priority' => $priority ?: 1024,
    ])->execute();
    $tableSchema = $this->db->getTableSchema($this->tableName);
    return $this->db->getLastInsertID($tableSchema->sequenceName);
}

return [
    'bootstrap' => [
        'queue', // 把这个组件注册到控制台
    ],
    'components' => [
        'db' => [
            'class' => \yii\db\Connection::class, 
            // ...
        ],
        'queue' => [
            'class' => \yii\queue\db\Queue::class,
            'db' => 'db', // DB 连接组件或它的配置
            'tableName' => '', // 表名
            'channel' => 'default', // Queue channel key
            'mutex' => \yii\mutex\MysqlMutex::class, // Mutex that used to sync queries
            'as log' => \yii\queue\LogBehavior::class,
        ],
    ],
];

CREATE TABLE `queue` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `channel` varchar(255) NOT NULL,
  `job` blob NOT NULL,
  `pushed_at` int(11) NOT NULL,
  `ttr` int(11) NOT NULL,
  `delay` int(11) NOT NULL DEFAULT 0,
  `priority` int(11) unsigned NOT NULL DEFAULT 1024,
  `reserved_at` int(11) DEFAULT NULL,
  `attempt` int(11) DEFAULT NULL,
  `done_at` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`),
  KEY `channel` (`channel`),
  KEY `reserved_at` (`reserved_at`),
  KEY `priority` (`priority`)
) ENGINE=InnoDB;

<?php
/**
 * @link http://www.yiiframework.com/
 * @copyright Copyright (c) 2008 Yii Software LLC
 * @license http://www.yiiframework.com/license/
 */

namespace yii\queue\db;

use yii\console\Exception;
use yii\queue\cli\Command as CliCommand;

/**
 * Manages application db-queue.
 *
 * @author Roman Zhuravlev <zhuravljov@gmail.com>
 */
class Command extends CliCommand
{
    /**
     * @var Queue
     */
    public $queue;
    /**
     * @var string
     */
    public $defaultAction = 'info';


    /**
     * @inheritdoc
     */
    public function actions()
    {
        return [
            'info' => InfoAction::class,
        ];
    }

    /**
     * @inheritdoc
     */
    protected function isWorkerAction($actionID)
    {
        return in_array($actionID, ['run', 'listen'], true);
    }

    /**
     * Runs all jobs from db-queue.
     * It can be used as cron job.
     *
     * @return null|int exit code.
     */
    public function actionRun()
    {
        return $this->queue->run(false);
    }

    /**
     * Listens db-queue and runs new jobs.
     * It can be used as daemon process.
     *
     * @param int $timeout number of seconds to sleep before next reading of the queue.
     * @throws Exception when params are invalid.
     * @return null|int exit code.
     */
    public function actionListen($timeout = 3)
    {
        if (!is_numeric($timeout)) {
            throw new Exception('Timeout must be numeric.');
        }
        if ($timeout < 1) {
            throw new Exception('Timeout must be greater than zero.');
        }

        return $this->queue->run(true, $timeout);
    }

    /**
     * Clears the queue.
     *
     * @since 2.0.1
     */
    public function actionClear()
    {
        if ($this->confirm('Are you sure?')) {
            $this->queue->clear();
        }
    }

    /**
     * Removes a job by id.
     *
     * @param int $id
     * @throws Exception when the job is not found.
     * @since 2.0.1
     */
    public function actionRemove($id)
    {
        if (!$this->queue->remove($id)) {
            throw new Exception('The job is not found.');
        }
    }
}

public $commandClass = Command::class;

public function bootstrap($app)
{
    if ($app instanceof ConsoleApp) {
        $app->controllerMap[$this->getCommandId()] = [
            'class' => $this->commandClass,
            'queue' => $this,  // Command类配置属性queue
        ] + $this->commandOptions;
    }
}

protected function runWorker(callable $handler)
{
    $this->_workerPid = getmypid();
    /** @var LoopInterface $loop */
    $loop = Yii::createObject($this->loopConfig, [$this]);

    $event = new WorkerEvent(['loop' => $loop]);
    $this->trigger(self::EVENT_WORKER_START, $event);
    if ($event->exitCode !== null) {
        return $event->exitCode;
    }

    $exitCode = null;
    try {
        call_user_func($handler, function () use ($loop, $event) {
            $this->trigger(self::EVENT_WORKER_LOOP, $event);
            return $event->exitCode === null && $loop->canContinue();
        });
    } finally {
        $this->trigger(self::EVENT_WORKER_STOP, $event);
        $this->_workerPid = null;
    }

    return $event->exitCode;
}