Yii Queue Extension

An extension for running tasks asynchronously via queues.

Documentation is at docs/guide/README.md.

Installation

The preferred way to install this extension is through composer.

Either run

php composer.phar require --prefer-dist yiisoft/yii-queue

or add

"yiisoft/yii-queue": "~3.0"

to the require section of your composer.json file.

Ready for yiisoft/config

If you are using yiisoft/config, you'll find out this package has some defaults in the common and params configurations saving your time. Things you should change to start working with the queue:

Optionally: define default \Yiisoft\Yii\Queue\Adapter\AdapterInterface implementation.
And/or define channel-specific AdapterInterface implementations in the channel-definitions params key to be used with the queue factory.
Define message handlers in the handlers params key to be used with the QueueWorker.
Resolve other \Yiisoft\Yii\Queue\Queue dependencies (psr-compliant event dispatcher).

Differences to yii2-queue

If you have experience with yiisoft/yii2-queue, you will find out that this package is similar. Though, there are some key differences which are described in the "migrating from yii2-queue" article.

Basic Usage

Each queue task consists of two parts:

A message is a class implementing MessageInterface. For simple cases you can use the default implementation, Yiisoft\Yii\Queue\Message\Message. For more complex cases you should implement the interface by your own.
A message handler is a callable called by a Yiisoft\Yii\Queue\Worker\Worker. The handler handles each queue message.

For example, if you need to download and save a file, your message may look like the following:

$data = [
    'url' => $url,
    'destinationFile' => $filename,
];
$message = new \Yiisoft\Yii\Queue\Message\Message('file-download', $data);

Then you should push it to the queue:

$queue->push($message);

Its handler may look like the following:

class FileDownloader
{
    private string $absolutePath;

    public function __construct(string $absolutePath) 
    {
        $this->absolutePath = $absolutePath;
    }

    public function handle(\Yiisoft\Yii\Queue\Message\MessageInterface $downloadMessage): void
    {
        $fileName = $downloadMessage->getData()['destinationFile'];
        $path = "$this->absolutePath/$fileName"; 
        file_put_contents($path, file_get_contents($downloadMessage->getData()['url']));
    }
}

The last thing we should do is to create a configuration for the Yiisoft\Yii\Queue\Worker\Worker:

$handlers = ['file-download' => [new FileDownloader('/path/to/save/files'), 'handle']];
$worker = new \Yiisoft\Yii\Queue\Worker\Worker(
    $handlers, // Here it is
    $logger,
    $injector,
    $container
);

There is the way to run all the messages that are already in the queue, and then exit:

$queue->run(); // this will execute all the existing messages
$queue->run(10); // while this will execute only 10 messages as a maximum before exit

If you don't want your script to exit immediately, you can use the listen method:

$queue->listen();

You can also check the status of a pushed message (the queue adapter you are using must support this feature):

$queue->push($message);
$id = $message->getId();

// Get status of the job
$status = $queue->status($id);

// Check whether the job is waiting for execution.
$status->isWaiting();

// Check whether a worker got the job from the queue and executes it.
$status->isReserved();

// Check whether a worker has executed the job.
$status->isDone();

Different queue channels

Often we need to push to different queue channels with an only application. There is the QueueFactory class to make different Queue objects creation for different channels. With this factory channel-specific Queue creation is as simple as

$queue = $factory->get('channel-name');

The main usage strategy is with explicit definition of channel-specific adapters. Definitions are passed in the $definitions constructor parameter of the factory, where keys are channel names and values are definitions for the Yiisoft\Factory\Factory. Below are some examples:

use Yiisoft\Yii\Queue\Adapter\SynchronousAdapter;

[
    'channel1' => new SynchronousAdapter(),
    'channel2' => static fn(SynchronousAdapter $adapter) => $adapter->withChannel('channel2'),
    'channel3' => [
        'class' => SynchronousAdapter::class,
        '__constructor' => ['channel' => 'channel3'],
    ],
]

For more information about a definition formats available see the factory documentation.

Another queue factory usage strategy is implicit adapter creation via withChannel() method call. To use this approach you should pass some specific constructor parameters:

true to the $enableRuntimeChannelDefinition
a default AdapterInterface implementation to the $defaultAdapter.

In this case $factory->get('channel-name') call will be converted to $this->queue->withAdapter($this->defaultAdapter->withChannel($channel)), when there is no explicit adapter definition in the $definitions.

Warning: This strategy is not recommended as it does not give you any protection against typos and mistakes in channel names.

Adapter behaviors

Some of queue adapters support different behaviors like delayed execution and prioritized queues.

Important: Not every adapter supports all the behaviors. See concrete adapter documentation to find out if it supports the behavior you want to use. Adapter will throw a BehaviorNotSupportedException if it does not support some behaviors attached to the message you are trying to push to the queue.

Delay behavior

To be sure message will be read from queue not earlier then 5 seconds since it will be pushed to the queue, you can use DelayBehavior:

$message->attachBehavior(new DelayBehavior(5));
$queue->push($message);

Console execution

The exact way of task execution depends on the adapter used. Most adapters can be run using console commands, which the component automatically registers in your application.

The following command obtains and executes tasks in a loop until the queue is empty:

yii queue/run

The following command launches a daemon which infinitely queries the queue:

yii queue/listen

See the documentation for more details about adapter specific console commands and their options.

The component also has the ability to track the status of a job which was pushed into queue.

For more details see the guide.

Unit testing

The package is tested with PHPUnit. To run tests:

./vendor/bin/phpunit

Mutation testing

The package tests are checked with Infection mutation framework. To run it:

./vendor/bin/infection

Static analysis

The code is statically analyzed with Psalm. To run static analysis:

./vendor/bin/psalm

Support the project

Follow updates

License

The Yii Queue Extension is free software. It is released under the terms of the BSD License. Please see LICENSE for more information.

Maintained by Yii Software.

romkatsu/yii-queue