Yii2 使用 Beanstalk 队列

安装

安装服务端

根据官方的教程安装，一条命令搞定。

安装 Yii2 客户端扩展

在你 Yii2 项目中执行

$ composer require udokmeci/yii2-beanstalk "^1.0.0"

安装管理 Beanstalk 队列的 Web 系统

此项目是一个单独的系统，可以查看和管理队列任务

$ composer create-project ptrofimov/beanstalk_console -s dev path/to/install

使用

开启服务端

$ beanstalkd -l 127.0.0.1 -p 11347 &

Yii2 端使用

添加配置文件：

'components' => [
    // ...
    'beanstalk' => [
        'class' => 'udokmeci\yii2beanstalk\Beanstalk',
        'host' => 127.0.0.1, // default host
        'port' => 11347,
        'connectTimeout' => 1,
        'sleep' => false, // or int for usleep after every job
    ],
    // ...
],

console 配置文件添加如下配置：

'controllerMap' => [
    'worker'=>[
        'class' => 'app\commands\WorkerController',
    ]
],

生产者，往队列里面添加数据示例，第二个参数，就是执行的时候需要的数据：

\Yii::$app->beanstalk->putInTube('track', ['unique_id' => 11, 'source' => 2, 'method'=> 'get']);

执行者，执行队列里的数据，在 app\commands\WorkerController 里面：

<?php
namespace app\commands;

use common\helpers\ObjectHelper;
use common\models\MongoTrack;
use common\services\TrackService;
use Pheanstalk\Job;
use udokmeci\yii2beanstalk\BeanstalkController;
use Yii;

class WorkerController extends BeanstalkController
{
    /**
     * @var integer 优先级 值越小优先级越高，默认为1024
     */
    const DELAY_PRIORITY = 1000;

    /**
     * @var integer 延迟执行的秒数
     */
    const DELAY_TIME = 5;

    /**
     * @var  integer 当 delay_max 达到 删除或延迟
     */
    const DELAY_MAX = 3;

    public function listenTubes()
    {
        return ['tube', 'track'];
    }


    /**
     * 采集数据
     * @param $job Job
     * @return string
     */
    public function actionTrack($job)
    {
        /** @var object $sentData */
        $sentData = $job->getData();
        if (TrackService::isNeedTrack($sentData->unique_id, $sentData->source, $sentData->method)) {
            $model = new MongoTrack();
            $model->load(ObjectHelper::toArray($sentData), '');
            fwrite(STDOUT, json_encode($model->errors));
            if ($model->save()) {
                return self::DELETE;
            }
            fwrite(STDOUT, json_encode($model->errors));
            return self::BURY;
        }
        // 不需要保存 mongodb 删除
        return self::DELETE;
    }

    // ...
}

最后记得要执行你写的 Worker：

$ php yii worker 

参考文献

• Beanstalk
• 中文文档
• Beanstalk console
• Yii2-beanstalk
• beanstalkd常见问题解答

本文由 forecho 创作，采用 知识共享署名 3.0 中国大陆许可协议 进行许可。 可自由转载、引用，但需署名作者且注明文章出处。

打赏作者

如果这篇文章对您有帮助，不妨微信小额赞助我一下，让我有动力继续写出高质量的教程。