最近在工作中使用队列碰到了一个问题,就是更新了队列的代码,线上的队列并没有运用到最新的代码,原因是旧的队列进程没有退出操作,一直在运行。而没有退出操作的队列,是不会被 Supervisor 管理到的,只能手动杀死。总不能一更新队列代码,就劳烦运维杀死旧的进程吧?
实际项目里要用队列的话已经有前人写好的 shell 脚本了,是下面这样子的。
#!/bin/bash php=$(which php) # $() 是将括号内命令的执行结果赋值给变量 if [ $php = "" ];then php="/usr/bin/php" fi pwdDir=$(cd `dirname $0`; pwd) # `` 表示先执行 `` 中的命令,dirname 表示去除文件名中的非目录部分 baseDir=$(dirname $pwdDir) runPHP=$baseDir"/public/index.php " while true do $php $runPHP "/$1/$2/$3" done
关键的代码是下面的那个 while 循环,每次执行完队列进程后,又会重启一个新的进程。这样新的进程一直可以使用到新的代码。而业务中就不需要再通过死循环取数据了,直接取一定量的数据执行,执行完后退出即可。
业务中可以执行这么取数据,第二个参数是每次取的条数。
$data = $queue->batchPop('queueKey', 100);
foreach ($data as $item) {
// 业务处理
}
下面是项目中封装的批量取 redis 队列的方法。
/**
* 从队列头部取出一个数据
*
* @param string $key 要取出队列名称
* @param bool $block 是否阻塞式读取
* @param int $timeout 超时时间,单位秒
* @return array $data 从队列里取出的数据
*/
public function pop(string $key, bool $block = true, $timeout = 5)
{
$this->ping();
if ($block) {
$data = $this->_queue->blPop($key, $timeout);
if (is_array($data) && !empty($data) && count($data) == 2) {
list($queueName, $queueData) = $data;
return is_numeric($queueData) ? $queueData : json_decode($queueData, true);
} else {
return false;
}
} else {
$queueData = $this->_queue->lPop($key);
if ($queueData !== false) {
return is_numeric($queueData) ? $queueData : json_decode($queueData, true);
} else {
return false;
}
}
}
/**
* 批量从队列头部取出多个数据
* @param string $key 要取出队列名称
* @param int $len 要取出队列的长度
* @param bool $block 是否阻塞式读取
* @param int $timeout 超时时间,单位秒
* @return array $data 从队列里取出的数据
*/
public function batchPop(string $key, int $len = 1, bool $block = true, $timeout = 1)
{
$this->ping();
$data = [];
for ($i = 0; $i < $len; $i++) {
$_data = $this->pop($key, $block, $timeout);
if ($_data === false) {
return $data;
} else {
$data[] = $_data;
}
}
return $data;
}