队列扩展, 支持多个队列

<?php

/**
* @author 魔芋红茶
*
*/
class MsgQuery {
// TODO - Insert your code here
private static $KEY_CACHE_PREFIX = 'mass.query.cache'; // 消息缓冲key前缀
private static $KEY_QUERY_PREFIX = 'mass.query.lv'; // 消息key
private static $KEY_CACHE_DEAL_PREFIX = 'mass.query.deal'; // 已处理缓冲key前缀
const SCORE_NUM = 5; // 优先级划分数目
const MIN_SCORE = 1; // 最小优先级
private static $MAX_SCORE;
private static $instance = null;
private $redis;
private $curQuery;
const QUERY_DEFAULT           = 'ms.q.dft';
const QUERY_ITEM_TASK            = 'ms.q.itm';
const QUERY_ITEM_SHELVES_ADD  = 'ms.q.shelves_add';  // 上架
const QUERY_ITEM_SHELVES_MOVE = 'ms.q.shelves_move'; // 下架
const QUERY_ITEM_IMPORT          = 'ms.q.import';          // 导入
const QUERY_ITEM_PRICE           = 'ms.q.price';          // 修改售价
const QUERY_ITEM_INV_QTY      = 'ms.q.inv_qty';      // 修改库存
const QUERY_ITEM_POINT            = 'ms.q.sell_point';      // 修改卖点
const QUERY_ITEM_DEL           = 'ms.q.itm.delete';      // 商品删除
const QUERY_ITEM_RESTORE      = 'ms.q.itm.restore';  // 商品恢复
const QUERY_ITEM_EDIT           = 'ms.q.itm.edit';      // 商品修改
const QUERY_ITEM_ADD           = 'ms.q.itm.add';      // 商品添加

private static $querys = array (
self::QUERY_DEFAULT,
self::QUERY_ITEM_TASK,
self::QUERY_ITEM_SHELVES_ADD,
self::QUERY_ITEM_SHELVES_MOVE,
self::QUERY_ITEM_IMPORT,
self::QUERY_ITEM_PRICE,
self::QUERY_ITEM_INV_QTY,
self::QUERY_ITEM_POINT,
self::QUERY_ITEM_DEL,
self::QUERY_ITEM_RESTORE,
self::QUERY_ITEM_ADD,
self::QUERY_ITEM_EDIT
);

/**
* 清理已经过期的query数据
*/
public function clean() {
$ystDay = date ( 'Ymd', strtotime ( '-1 day' ) );
$cacheKey = self::$KEY_CACHE_PREFIX . $ystDay;
$cacheDealKey = self::$KEY_CACHE_DEAL_PREFIX . $ystDay;
// 清理前尝试移动缓冲区里的残留消息到队列
$this->moveToQuery ( $ystDay );
// 删除过期缓冲区
$this->redis->del ( $cacheKey );
$this->redis->del ( $cacheDealKey );
}

public static function getQuerys() {
return self::$querys;
}

/**
* 将当前query切换到指定query
*
* @param string $query
*/
public function selectQuery($query) {
if (! in_array ( $query, self::$querys )) {
$query = self::QUERY_DEFAULT;
}
$this->curQuery = $query;
self::$KEY_CACHE_PREFIX = $query . '.cache';
self::$KEY_QUERY_PREFIX = $query . '.lv';
self::$KEY_CACHE_DEAL_PREFIX = $query . '.deal';
}

/**
* 获取一个query实例
* @param redis $redis
* @param string $query
* @return MsgQuery
*/
public static function getInstance($redis, $query) {
if (null == self::$instance) {
self::$instance = new MsgQuery ( $redis, $query );
}
if (self::$instance->curQuery != $query) {
self::$instance->selectQuery ( $query );
}
return self::$instance;
}

/**
* 添加消息到消息缓冲区
*
* @param int $score
*            优先级(1-5)
* @param string $msg
*            消息
*/
public function add($score, $msg) {
// 添加到消息缓冲
$socre = intval ( $score );
if ($socre < self::MIN_SCORE) {
$score = self::MIN_SCORE;
}
if ($score > self::$MAX_SCORE) {
$score = self::$MAX_SCORE;
}
$cacheKey = self::$KEY_CACHE_PREFIX . date ( 'Ymd' );
$cacheData = array (
'score' => $score,
'msg' => $msg
);

// 被添加到集合中的新元素的数量，不包括被忽略的元素，故重复添加返回false
return $this->redis->sAdd ( $cacheKey, serialize ( $cacheData ) );
}

/**
* 将消息从缓冲区移动到相应的优先级队列中
*/
public function moveToQuery($day = null) {
if ($day === null) {
$day = date ( 'Ymd' );
}
// 获取当前缓冲区没有入队列的消息
$dealKey = self::$KEY_CACHE_DEAL_PREFIX . $day;
$cacheKey = self::$KEY_CACHE_PREFIX . $day;
$msgs = $this->redis->sDiff ( $cacheKey, $dealKey );
foreach ( $msgs as $cachedData ) {
// 放入已处理集合
$this->redis->sAdd ( $dealKey, $cachedData );
// 压入相应的优先级队列
$cachedData = unserialize ( $cachedData );
$score = $cachedData ['score'];
$msg = $cachedData ['msg'];
$queryKey = self::$KEY_QUERY_PREFIX . $score;
$this->redis->rPush ( $queryKey, serialize($msg) );
}
unset ( $cachedData );
}

/**
* 从队列阻塞式出栈一个最高优先级消息
*
* @return string msg
*/
public function bPop() {
$queryKeys = array ();
for($score = self::$MAX_SCORE; $score >= self::MIN_SCORE; $score --) {
$queryKeys [] = self::$KEY_QUERY_PREFIX . $score;
}
$msg = @$this->redis->blPop ( $queryKeys, 0 );
return $msg [1];
}

private function __construct($redis, $query) {
$this->redis = $redis;
//         $this->redis->connect ();
self::$MAX_SCORE = self::MIN_SCORE + self::SCORE_NUM - 1;
$this->selectQuery ( $query );
}

public function __destruct() {
if ($this->redis)     $this->redis->close ();
}
}

?>

主体没有任何改变, 依然是2个set做缓冲, 多个list做实际队列的实现, 只是扩展为支持多个队列的实现.

其它队列使用中遇到过的问题:

1 feed程序因为长时间阻塞而断开了和redis的连接

解决方法: 加入代码

ini_set('default_socket_timeout', -1);

以上代码可以让php进程和redis通过socket长连接

2 feed程序因为长时间阻塞而断开了和mysql的连接

解决方法: 查看mysql驱动使用的是哪种, 我们项目中用的是mysqli连接, 具体驱动是mysqlnd, 如果是mysqld, 最简单的方式是先设置php变量mysqli.reconnect=1再加入ping命令

但有个问题是mysqlnd驱动下ping命令并不能自动重连, 即使改了php变量也没用, 只能显示进行重连, 如下

if(!$this->connection->ping()){
//mysql 连接丢失且mysqlnd不会自动重连, 手动重连
$this->connectDB();
}

3 重新发布/变更了feed程序可能需要重启进程加载新代码

最后附上一个队列分发和清理任务脚本, 脚本中多进程分发的代码存在问题, 会产生很多僵尸进程, 目前仅使用单进程版本, 处理时间为每天凌晨调用clean脚本清理前一天的队列, 分发程序为每5分钟执行一次, 具体执行间隔视情况调整

<?php
class ControllerMsgQuery extends Controller {

/**
* 处理队列缓冲区的数据, 移动到队列
*/
public function index() {
//多进程版本存在问题, 先使用单进程版本
$this->dealWithSingleProcess();
}

private function dealWithMuitlProcess(){
set_time_limit ( 0 );
global $global;
$this->load->ventor ( 'msg_query/MsgQuery' );
$querys = MsgQuery::getQuerys ();
$redis = new Redis ();
$redis->connect ( $global['redis_cache_w'][0]['host'], $global['redis_cache_w'][0]['port'] );
foreach ( $querys as $query ) {
// 每个query用独立子线程完成move to list的工作
$pid = pcntl_fork ();
if ($pid > 0) {
// 父进程
} else if ($pid == 0) {
// 子进程
$MsgQuery = MsgQuery::getInstance ( $redis, $query );
$MsgQuery->moveToQuery ();
exit();
} else {
exit();
}
}
// 父进程等待子进程以释放资源
//         if ($pid > 0){
pcntl_wait ( $status );
//         }
}

private function dealWithSingleProcess(){
set_time_limit ( 0 );
global $global;
$this->load->ventor ( 'msg_query/MsgQuery' );
$querys = MsgQuery::getQuerys ();
$redis = new Redis ();
$redis->connect ( $global['redis_cache_w'][0]['host'], $global['redis_cache_w'][0]['port'] );
foreach ( $querys as $query ) {
$MsgQuery = MsgQuery::getInstance ( $redis, $query );
$MsgQuery->moveToQuery ();
}
}

/**
* 清理已经过期的query数据
*/
public function clean() {
set_time_limit ( 0 );
global $global;
$this->load->ventor ( 'msg_query/MsgQuery' );
$querys = MsgQuery::getQuerys ();
$redis = new Redis ();
$redis->connect ( $global['redis_cache_w'][0]['host'], $global['redis_cache_w'][0]['port'] );
foreach ( $querys as $query ) {
$MsgQuery = MsgQuery::getInstance ( $redis, $query );
$MsgQuery->clean ();
}
}
}