php的消息队列框架resque的使用小结

前段时间在开发一个量化交易系统，这是一个类似股票交易软件的系统：股票价格变化后要实时在终端（APP）的行情价格页面实时更新，如果用户是在持仓页面，还要计算持仓盈亏及预付款比例，因为有一个强制平仓机制（当预付款比较低于30%时系统要自动触发强制平仓），因为行情波动非常快（有些产品一秒钟内价格会变化五六次），所以当某个产品的价格变化时，要触发一系统的行情推送（通过长连接）及盈亏计算操作。

因为价格变动非常快，如果使用同步操作的话，很容易造成阻塞，所以用异步消息队列是比较好的选择，在网上查了一番，决定用resque来实现，试用了一下，发现网上所谓的resque教程大都是将它的demo代码讲解了一遍，坑很多，所以把我试用过程中遇到的问题统一做个记录，一个是方便同样正在尝试使用resque的朋友，二是作为自己的备忘录，好，下面正式开始。

Part 1.安装

网上很多教程的指引还是安装源chrisboulton/php-resque版本的，实际上现在最新的版本应该是resque/php-resque源的，用composer安装的方法是，在composer.json文件中的require节点下添加”resque/php-resque:^1.2”，

然后执行composer install即可。

Part2.连接设置了密码的redis

resque是基于redis的，所以没有redis是运行不起来的，可能开发者初期没有考虑到redis有设置了密码的情况，所以我刚开始看网上的资料是有介绍怎样解决这个问题的，但我把代码签出来后发现Resque::setBackend的方法是支持传入auth参数的，于是以为这个问题已经在目前的版本里解决了，实际上还是不行，如果你的redis里设置了密码，建议修改插件目录下的resque/php-resque/lib/Resque.php这个文件，大概在68行将创建redis连接的代码用以下代码代替可以解决连接redis失败的问题。

代码语言：javascript复制

$redis = new Redis();
$redis->connect(self::$redisServer);
$redis->auth(self::$auth);
$redis->select(self::$redisDatabase);
self::$redis = $redis;

修改完后将redis的密码通过setBackend方法的第三个参数传入就可以了，示例：

代码语言：javascript复制

Resque::setBackend('127.0.0.1', 1, Config::$redis['option']['auth']);

Part3.运行

如果你看了其它网友的经验分享，你会知道resque有三个角色:Queue、Worker、Job，Queue负责接收消息队列，Worker负责任务调度，Job负责执行业务逻辑，用现实生活举例就是像一个高铁站一样，Queue是售票厅、Worker是控制室、Job则是车队，以下我也分三部分说下这个注意事项：

Queue: 按照业务逻辑需要，Queue在何时何地执行都可以的，因为作为一个消息队列，肯定要支持动态添加任务进去，有一点要注意的是，如果你的Job类有命名空间，一定要将完整的路径写进去，我在这里卡了一下，以为在Job类里通过use引入了这个命名空间即可，但实际上行不通，必须在调用enqueue方法时传入带命名空间的路径才能正常执行，下面是一个示例：

代码语言：javascript复制

Resque::setBackend('127.0.0.1', 1, Config::$redis['option']['auth']);
for($i = 1; $i < 10; $i  ){
    Resque::enqueue('pushMarketPrice', 'site_pcactionJob', ['newPrice' => 1367.85, 'rate' => 7.09]);
}

Worker 从demo文件可以看出，Worker就是要加载Job类及bin下的resque文件，但demo里并不是面向对象的写法，如果要定义在一个类里，可以参照以下写法：

代码语言：javascript复制

class Worker{
    /**
     * @router cli work
     */
    public function run(){
        require "../vendor/resque/php-resque/bin/resque";
    }
}

其中，require后面的路径请修改为相对你项目执行文件入口的路径。

Job Job类就比较简单，记着要实现一个perform方法就方法了，另外，demo里没有体现的一个细节是在Queue中传入的参数怎样获取，其实很简单，在perform方法通过$this->args就能拿到传入的参数数组，非常方便。

代码语言：javascript复制

class Job{
    public function perform(){
        Tools::logToDb('resque', $this->args);
    }
}

$this->args就是你此前在Queue中传入的自定义参数数组了。

Part4.监控

怎样判断resque的状态（是否在正常运行中）呢？还有如果已经在运行，怎样重启？可以通过查看进程的命令来达到这个目的，登录系统后执行： ps -aux | grep resque 如果resque正在运行中，你看到的界面应该是这样的：

其中，上图椭圆形圈中的是进程ID，如果希望停止运行，可以使用kill -9 [进程id]来杀死进程，红方框则是进程状态，正在等待队列中的任务。一有任务就会执行，至此，这个插件已经在正常运行了。

云数据库Redis php composer 消息队列CMQ版 kafka

0 人点赞