---

Итак у нас есть модель Good - описание товара, с ценой и фотографиями. Определим события, которые требует Заказчик.

class Good extends ActiveRecord
{
    const EVENT_NEW_PICTURE         = 'new_picture';
    const EVENT_DESCRIPTION_CHANGED = 'desciption_changed';
    const EVENT_PRICE_REDUCTION     = 'price_reduction';
    /**
     * @inheritdoc
     */
    public function init()
    {
        parent::init();
        $this->on(self::EVENT_NEW_PICTURE, function($event) {
            $this->notifyObservers($event);
        });
        $this->on(self::EVENT_DESCRIPTION_CHANGED, function($event) {
            $this->notifyObservers($event);
        });
        $this->on(self::EVENT_PRICE_REDUCTION, function($event) {
            $this->notifyObservers($event);
        });
    }
    /**
     * Get observers
     * 
     * @return \yii\db\ActiveRecord
     */
    public function getObservers()
    {
        return $this->hasMany(Wishlist::className(), [
            'good_id' => 'id',
        ]);
    }
    /**
     * Notify observers
     *     
     * @param \yii\base\Event $event
     */
    public function notifyObservers($event)
    {
        foreach($this->observers as $observer) {
            $this->keepNotification([
                'user_id' => $observer->user_id,
                'good_id' => $this->id,
                'event'   => $event->name,
            ]);
        }
    }
    /**
     * Keep information about an event in a list
     * 
     * @param array $data
     */
    public function keepNotification($data)
    {
        $file = fopen(Yii::$app->queue->path . '/data.csv', 'a');
        fwrite($file, "{$data['user_id']},{$data['good_id']},{$data['event']}\n");
        fclose($file);
    }

Обрабатываются события в два этапа. Сначала события сохраняются в общем списке (используется обычный файл). События сохраняются одно за другим до момента старта воркера (обработчика очереди).

php yii queue/run -v --color

При старте воркера будет вызван обработчик события Queue::EVENT_WORKER_START, который формирует задания, по одному на каждого юзера и толкает их в очередь. Это второй этап.

class PrepareNotificationBehavior extends Behavior
{
    public function events()
    {
        return [
            Queue::EVENT_WORKER_START => 'prepareNotification',
        ];
    }
    public function prepareNotification($event)
    {
        $queue = $event->sender;
        if (!file_exists($queue->path . '/data.csv'))
            // no events
            return;
        $jobs = [];
        // spread out all events by users, then by goods
        // for example user_id = 13, good_id = 1001, then 
        // $jobs[13][1001] = ['new_picture', 'price_reduction'];
        foreach(file($queue->path . '/data.csv') as $line) {
            if (count($a = str_getcsv($line)) == 3) {
                list($user_id, $good_id, $event_name) = $a;
                if (!isset($jobs[$user_id = (int) $user_id]))
                    $jobs[$user_id] = [];
                if (!isset($jobs[$user_id][$good_id = (int) $good_id]))
                    $jobs[$user_id][$good_id] = [];
                if (!isset($jobs[$user_id][$good_id][$event_name = trim($event_name)]))
                    $jobs[$user_id][$good_id][] = $event_name;
            }
        };
        // create jobs, one for each users
        foreach($jobs as $user_id => $goods) {
            $queue->push(new SendNotificationJob([
                'user_id' => $user_id,
                'goods'   => $goods,
            ]));
        }
        // delete event list
        unlink($queue->path . '/data.csv');
    }
}

Обработчик надо подключить к очереди через config.

    'components' => [ 
        'queue' => [ 
            'class' => \yii\queue\file\Queue::class, 
            'path' => '@console/runtime/queue', 
            'as prep' => \common\behaviors\PrepareNotificationBehavior::class, 
        ],

Так как задания формируются перед началом работы воркера, они тут же и выполняются.

class SendNotificationJob extends BaseObject implements \yii\queue\JobInterface
{
    public $user_id;
    public $goods;
    /**
     * Send email to observer
     */
    public function execute($queue)
    {
        // find User
        if (!($user = User::findOne(['id' => $this->user_id])))
            throw new NotFoundHttpException('User with ID - ' . 
            $this->user_id . ' not exists.');
        // generate and save token, for unsubscribe link
        if (!User::isPasswordResetTokenValid($user->password_reset_token)) {
            $user->generatePasswordResetToken();
            $user->save();
        }
        // find goods
        $models = Goods::find()
            ->where(['in', 'id', array_keys($this->goods)])
            ->all();
        // send email
        if ($this->goods && $models)
            Yii::$app->mailer
                ->compose(['html' => 'notification-html'], [
                    'user'   => $user, 
                    'models' => $models, 
                    'goods'  => $this->goods, 
                ])
                ->setFrom([Yii::$app->params['email']['support'] => 
                    (Yii::$app->name . ' robot')])
                ->setTo($user->email)
                ->setSubject($subject)
                ->send();
    }
}

Заключение

Запуск вокера предполагается выполнять только командой run руками или через cron. Если в приложении используется очередь и для других целей, то стоит определить дополнительную очередь в конфигурации.