php 封装Memcache 队列缓存类

/**
* ----扩展redis列表<a target=_blank href="http://qifuguang.me/2015/09/29/Redis%E4%BA%94%E7%A7%8D%E6%95%B0%E6%8D%AE%E7%B1%BB%E5%9E%8B%E4%BB%8B%E7%BB%8D/" target="_blank">http://qifuguang.me/2015/09/29/Redis%E4%BA%94%E7%A7%8D%E6%95%B0%E6%8D%AE%E7%B1%BB%E5%9E%8B%E4%BB%8B%E7%BB%8D/</a>------
*
* memcache:add($key,$val,$flag,$expire);
* $key:将要分配给变量的key
* $val:将要被存储的变量值
* $flag:MEMCACHE_COMPRESSED,标记对数据进行压缩（等于2）
* $expire:当写入缓存数据的失效时间
*/
/**
* Memcache缓存队列类
* @author    ricky
*/
class CacheMemcacheQueue{
public static $client;   //memcache客户端连接
public $access;   //队列是否可更新
private $expire;   //过期时间,秒,1~2592000,即30天内
private $sleepTime;   //等待解锁时间,微秒
private $queueName;   //队列名称,唯一值
private $retryNum;   //重试次数,= 10 * 理论并发数
public $currentHead;  //当前队首值
public $currentTail;  //当前队尾值

public $maxnum  = 20;    //最大队列数,建议上限10K
const HEAD_KEY = '_QueueHead_';  //队列首kye
const TAIL_KEY = '_QueueTail_';  //队列尾key
const VALU_KEY = '_QueueValu_';  //队列值key
const LOCK_KEY = '_QueueLock_';  //队列锁key

/**
* 构造函数
* @param string $queueName  队列名称
* @param int $expire       缓存队列生命周期时间
* @return <type>
*/
function __construct($queueName, $expire = 0, $config= '',$maxnum = 0) {
if(empty($config)){
self::$client = memcache_pconnect('127.0.0.1',11211);
}elseif(is_array($config)){//array('host'=>'127.0.0.1','port'=>'11211')
self::$client = memcache_pconnect($config['host'],$config['port']);
}elseif(is_string($config)){//"127.0.0.1:11211"
$tmp = explode(':',$config);
$conf['host'] = isset($tmp[0]) ? $tmp[0] : '127.0.0.1';
$conf['port'] = isset($tmp[1]) ? $tmp[1] : '11211';
self::$client = memcache_pconnect($conf['host'],$conf['port']);
}
if(!self::$client) return false;

ignore_user_abort(true); //当客户断开连接,允许继续执行
set_time_limit(0); //取消脚本执行延时上限
$this->access = false;//设置初始值，队列不可更新
$this->sleepTime = 1000;//设置等待解锁时间1秒
$expire = empty($expire) ? 3600 : intval($expire) + 1;
$this->expire = $expire;//设置过期时间
$this->queueName = $queueName;//设置队列名称
$this->retryNum = 1000;//设置重试次数
$this->maxnum = $maxnum?$maxnum:$this->maxnum;
$this->_initSetHeadNTail();//初始化
//设置队列首尾值用来统计队列是否已满
//         $this->currentTail - $this->currentHead >= $this->maxnum，说明队列已满

}
/**
* 初始化设置队列首尾值
*/
private function _initSetHeadNTail() {
//当前队列首的数值
$head_key = $this->queueName . self::HEAD_KEY;
$tail_key = $this->queueName . self::TAIL_KEY;
$this->currentHead = self::$client->get($head_key);
if ($this->currentHead === false)
$this->currentHead = 0;
//当前队列尾的数值
$this->currentTail = self::$client->get($tail_key);
if ($this->currentTail === false)
$this->currentTail = 0;
}
/**
* head_key,tail_key
*/
public function getHeadTailKey(){
$head_key = $this->queueName . self::HEAD_KEY;
$tail_key = $this->queueName . self::TAIL_KEY;
return 'head_Key:'.self::$client->get($head_key)
.'----'.'tail_key'.self::$client->get($tail_key);
}
/**
* 当取出元素时,改变队列首的数值
* @param int $step 步长值
*/
private function _changeHead($step = 1) {
$head_key = $this->queueName . self::HEAD_KEY;
$this->currentHead += $step;
self::$client->set($head_key, $this->currentHead, false, $this->expire);
}
/**
* 当添加元素时,改变队列尾的数值
* @param int $step 步长值
* @param bool $reverse 是否反向
* @return null
*/
private function _changeTail($step = 1, $reverse = false) {
$tail_key = $this->queueName . self::TAIL_KEY;
if (!$reverse) {
$this->currentTail += $step;
} else {
$this->currentTail -= $step;
}
self::$client->set($tail_key, $this->currentTail, false, $this->expire);
}
/**
* 队列是否为空
* @return bool
*/
private function _isEmpty() {
return (bool) ($this->currentHead === $this->currentTail);
}
/**
* 队列是否已满
* @return bool
*/
private function _isFull() {
$len = $this->currentTail - $this->currentHead;
return (bool) ($len === $this->maxnum);
}

/**
* 队列加锁
* $this->access = false：表示队列未锁住
*/
private function _getLock() {
$head_key = $this->queueName . self::HEAD_KEY;
$i = 0;
if ($this->access === false) {//如果队列可更新，队列首值增加1，同时，用while语句达到等待的效果
while ($result = self::$client->add($head_key, 1, MEMCACHE_COMPRESSED, $this->expire)) {
usleep($this->sleepTime);//等待解锁时间
$i++;
if ($i > $this->retryNum) {//尝试等待N次（设置重试次数）
return false;
break;
}
}
$this->_initSetHeadNTail();//初始化设置队列首尾值
return $this->access = true;//队列加锁
}
return $this->access;
}

/**
* 队列解锁
*/
private function _unLock() {
$lock_key = $this->queueName . self::VALU_KEY;
self::$client->delete($lock_key,0);//第二个参数
$this->access = false;
}

/**
* 添加队列数据
* @param void $data 要添加的数据
* @return bool
*/
public function queueAdd($data) {
if (!$this->_getLock())
return false;
if ($this->_isFull()) {//如果队列已满
$this->_unLock();//解锁
return false;//不允许回调
}
//检测数据是否写入
$list = $this->queueRead();
if(!empty($list) && in_array($data,$list)){
$this->_unLock();
return false;
}
$value_key = $this->queueName . self::VALU_KEY . strval($this->currentTail + 1);
$result = self::$client->set($value_key, $data, MEMCACHE_COMPRESSED, $this->expire);
if ($result) {//添加成功后
$this->_changeTail();//改变队列尾的数值
}
$this->_unLock();//解锁
return $result;
}
/**
* 读取队列数据
* @param int $length 要读取的长度(反向读取使用负数)
* @return array
*/
public function queueRead($length = 0) {
if (!is_numeric($length))
return false;
$this->_initSetHeadNTail();
if ($this->_isEmpty()) {
return false;
}
if (empty($length))
$length = $this->maxnum; //默认所有
$keyArr = array();
if ($length > 0) {//正向读取(从队列首向队列尾)
$tmpMin = $this->currentHead;
$tmpMax = $tmpMin + $length;
for ($i = $tmpMin; $i <= $tmpMax; $i++) {
$keyArr[] = $this->queueName . self::VALU_KEY . $i;
}
} else {//反向读取(从队列尾向队列首)
$tmpMax = $this->currentTail;
$tmpMin = $tmpMax + $length;
for ($i = $tmpMax; $i > $tmpMin; $i--) {
$keyArr[] = $this->queueName . self::VALU_KEY . $i;
}
}
$result = self::$client->get($keyArr);
return $result;
}

/**
* 取出队列数据
* @param int $length 要取出的长度(反向读取使用负数)
* @return array
*/
public function queueGet($length = 0) {
if (!is_numeric($length))
return false;
if (!$this->_getLock())
return false;

if ($this->_isEmpty()) {
$this->_unLock();
return false;
}
if (empty($length))
$length = $this->maxnum; //默认所有
$length = intval($length);
$keyArr = array();
if ($length > 0) {//正向读取(从队列首向队列尾)
$tmpMin = $this->currentHead;
$tmpMax = $tmpMin + $length;
for ($i = $tmpMin; $i <= $tmpMax; $i++) {
$keyArr[] = $this->queueName . self::VALU_KEY . $i;
}
$this->_changeHead($length);
} else {//反向读取(从队列尾向队列首)
$tmpMax = $this->currentTail;
$tmpMin = $tmpMax + $length;
for ($i = $tmpMax; $i > $tmpMin; $i--) {
$keyArr[] = $this->queueName . self::VALU_KEY . $i;
}
$this->_changeTail(abs($length), true);
}
$result = self::$client->get($keyArr);
foreach ($keyArr as $v) {//取出之后删除
self::$client->delete($v,0);
}
$this->_unLock();
return $result;
}
/**
* 清空队列
*/
public function queueClear() {
$head_key = $this->queueName . self::HEAD_KEY;
$tail_key = $this->queueName . self::TAIL_KEY;
if (!$this->_getLock())return false;

if ($this->_isEmpty()) {
$this->_unLock();
return false;
}
$tmpMin = $this->currentHead--;
$tmpMax = $this->currentTail++;

$tmpMax = $this->maxnum;//最大队列数

for ($i = $tmpMin; $i <= $tmpMax; $i++) {
$tmpKey = $this->queueName . self::VALU_KEY . $i;
//memcache_delete(self::$client, $tmpKey, 0);
self::$client->delete($tmpKey,0);
}
$this->currentTail = $this->currentHead = 0;
self::$client->set($head_key, $this->currentHead, false, $this->expire);
self::$client->set($tail_key, $this->currentTail, false, $this->expire);
$this->_unLock();
}

/*
* 清除所有memcache缓存数据
*/

public function memFlush() {
//memcache_flush(self::$client);
self::$client->flush();
}

/*
* 单一删除缓存
*/
public function queueDelete($key){
$head_key = $this->queueName . self::HEAD_KEY;
$tail_key = $this->queueName . self::TAIL_KEY;
if (!$this->_getLock())return false;

if ($this->_isEmpty()) {
$this->_unLock();
return false;
}
$tmpMin = $this->currentHead--;
$tmpMax = $this->currentTail++;
for ($i = $tmpMin; $i <= $tmpMax; $i++) {
$tmpKey = $key . $i;
self::$client->delete($tmpKey,0);
}
$this->currentTail = $this->currentHead = 0;
self::$client->set($head_key, $this->currentHead, false, $this->expire);
self::$client->set($tail_key, $this->currentTail, false, $this->expire);
$this->_unLock();
return true;
}

}