memcacheQueue.class.php
<?php /* * memcache队列类 * 支持多进程并发写入、读取 * 边写边读,AB面轮值替换 * @author lkk/lianq.net * @create on 9:25 2012-9-28 * * * @example: * $obj = new memcacheQueue('duilie'); * $obj->add('1asdf'); * $obj->getQueueLength(); * $obj->read(11); * $obj->get(8); */ class memcacheQueue{ public static $client; //memcache客户端连接 public $access; //队列是否可更新 private $currentSide; //当前轮值的队列面:A/B private $lastSide; //上一轮值的队列面:A/B private $sideAHead; //A面队首值 private $sideATail; //A面队尾值 private $sideBHead; //B面队首值 private $sideBTail; //B面队尾值 private $currentHead; //当前队首值 private $currentTail; //当前队尾值 private $lastHead; //上轮队首值 private $lastTail; //上轮队尾值 private $expire; //过期时间,秒,1~2592000,即30天内;0为永不过期 private $sleepTime; //等待解锁时间,微秒 private $queueName; //队列名称,唯一值 private $retryNum; //重试次数,= 10 * 理论并发数 const MAXNUM = 2000; //(单面)最大队列数,建议上限10K const HEAD_KEY = '_lkkQueueHead_'; //队列首kye const TAIL_KEY = '_lkkQueueTail_'; //队列尾key const VALU_KEY = '_lkkQueueValu_'; //队列值key const LOCK_KEY = '_lkkQueueLock_'; //队列锁key const SIDE_KEY = '_lkkQueueSide_'; //轮值面key /* * 构造函数 * @param [config] array memcache服务器参数 * @param [queueName] string 队列名称 * @param [expire] string 过期时间 * @return NULL */ public function __construct($queueName ='',$expire='',$config =''){ if(empty($config)){ self::$client = memcache_pconnect('localhost',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; $expire = (empty($expire) && $expire!=0) ? 3600 : (int)$expire; $this->expire = $expire; $this->queueName = $queueName; $this->retryNum = 10000; $side = memcache_add(self::$client, $queueName . self::SIDE_KEY, 'A',false, $expire); $this->getHeadNTail($queueName); if(!isset($this->sideAHead) || empty($this->sideAHead)) $this->sideAHead = 0; if(!isset($this->sideATail) || empty($this->sideATail)) $this->sideATail = 0; if(!isset($this->sideBHead) || empty($this->sideBHead)) $this->sideBHead = 0; if(!isset($this->sideBHead) || empty($this->sideBHead)) $this->sideBHead = 0; } /* * 获取队列首尾值 * @param [queueName] string 队列名称 * @return NULL */ private function getHeadNTail($queueName){ $this->sideAHead = (int)memcache_get(self::$client, $queueName.'A'. self::HEAD_KEY); $this->sideATail = (int)memcache_get(self::$client, $queueName.'A'. self::TAIL_KEY); $this->sideBHead = (int)memcache_get(self::$client, $queueName.'B'. self::HEAD_KEY); $this->sideBTail = (int)memcache_get(self::$client, $queueName.'B'. self::TAIL_KEY); } /* * 获取当前轮值的队列面 * @return string 队列面名称 */ public function getCurrentSide(){ $currentSide = memcache_get(self::$client, $this->queueName . self::SIDE_KEY); if($currentSide == 'A'){ $this->currentSide = 'A'; $this->lastSide = 'B'; $this->currentHead = $this->sideAHead; $this->currentTail = $this->sideATail; $this->lastHead = $this->sideBHead; $this->lastTail = $this->sideBTail; }else{ $this->currentSide = 'B'; $this->lastSide = 'A'; $this->currentHead = $this->sideBHead; $this->currentTail = $this->sideBTail; $this->lastHead = $this->sideAHead; $this->lastTail = $this->sideATail; } return $this->currentSide; } /* * 队列加锁 * @return boolean */ private function getLock(){ if($this->access === false){ while(!memcache_add(self::$client, $this->queueName .self::LOCK_KEY, 1, false, $this->expire) ){ usleep($this->sleepTime); @$i++; if($i > $this->retryNum){//尝试等待N次 return false; break; } } return $this->access = true; } return false; } /* * 队列解锁 * @return NULL */ private function unLock(){ memcache_delete(self::$client, $this->queueName .self::LOCK_KEY); $this->access = false; } /* * 添加数据 * @param [data] 要存储的值 * @return boolean */ public function add($data){ $result = false; if(!$this->getLock()){ return $result; } $this->getHeadNTail($this->queueName); $this->getCurrentSide(); if($this->isFull()){ $this->unLock(); return false; } if($this->currentTail < self::MAXNUM){ $value_key = $this->queueName .$this->currentSide . self::VALU_KEY . $this->currentTail; if(memcache_add(self::$client, $value_key, $data, false, $this->expire)){ $this->changeTail(); $result = true; } }else{//当前队列已满,更换轮值面 $this->unLock(); $this->changeCurrentSide(); return $this->add($data); } $this->unLock(); return $result; } /* * 取出数据 * @param [length] int 数据的长度 * @return array */ public function get($length=0){ if(!is_numeric($length)) return false; if(empty($length)) $length = self::MAXNUM * 2;//默认读取所有 if(!$this->getLock()) return false; if($this->isEmpty()){ $this->unLock(); return false; } $keyArray = $this->getKeyArray($length); $lastKey = $keyArray['lastKey']; $currentKey = $keyArray['currentKey']; $keys = $keyArray['keys']; $this->changeHead($this->lastSide,$lastKey); $this->changeHead($this->currentSide,$currentKey); $data = @memcache_get(self::$client, $keys); foreach($keys as $v){//取出之后删除 @memcache_delete(self::$client, $v, 0); } $this->unLock(); return $data; } /* * 读取数据 * @param [length] int 数据的长度 * @return array */ public function read($length=0){ if(!is_numeric($length)) return false; if(empty($length)) $length = self::MAXNUM * 2;//默认读取所有 $keyArray = $this->getKeyArray($length); $data = @memcache_get(self::$client, $keyArray['keys']); return $data; } /* * 获取队列某段长度的key数组 * @param [length] int 队列长度 * @return array */ private function getKeyArray($length){ $result = array('keys'=>array(),'lastKey'=>array(),'currentKey'=>array()); $this->getHeadNTail($this->queueName); $this->getCurrentSide(); if(empty($length)) return $result; //先取上一面的key $i = $result['lastKey'] = 0; for($i=0;$i<$length;$i++){ $result['lastKey'] = $this->lastHead + $i; if($result['lastKey'] >= $this->lastTail) break; $result['keys'][] = $this->queueName .$this->lastSide . self::VALU_KEY . $result['lastKey']; } //再取当前面的key $j = $length - $i; $k = $result['currentKey'] = 0; for($k=0;$k<$j;$k++){ $result['currentKey'] = $this->currentHead + $k; if($result['currentKey'] >= $this->currentTail) break; $result['keys'][] = $this->queueName .$this->currentSide . self::VALU_KEY . $result['currentKey']; } return $result; } /* * 更新当前轮值面队列尾的值 * @return NULL */ private function changeTail(){ $tail_key = $this->queueName .$this->currentSide . self::TAIL_KEY; memcache_add(self::$client, $tail_key, 0,false, $this->expire);//如果没有,则插入;有则false; //memcache_increment(self::$client, $tail_key, 1);//队列尾+1 $v = memcache_get(self::$client, $tail_key) +1; memcache_set(self::$client, $tail_key,$v,false,$this->expire); } /* * 更新队列首的值 * @param [side] string 要更新的面 * @param [headValue] int 队列首的值 * @return NULL */ private function changeHead($side,$headValue){ if($headValue < 1) return false; $head_key = $this->queueName .$side . self::HEAD_KEY; $tail_key = $this->queueName .$side . self::TAIL_KEY; $sideTail = memcache_get(self::$client, $tail_key); if($headValue < $sideTail){ memcache_set(self::$client, $head_key,$headValue+1,false,$this->expire); }elseif($headValue >= $sideTail){ $this->resetSide($side); } } /* * 重置队列面,即将该队列面的队首、队尾值置为0 * @param [side] string 要重置的面 * @return NULL */ private function resetSide($side){ $head_key = $this->queueName .$side . self::HEAD_KEY; $tail_key = $this->queueName .$side . self::TAIL_KEY; memcache_set(self::$client, $head_key,0,false,$this->expire); memcache_set(self::$client, $tail_key,0,false,$this->expire); } /* * 改变当前轮值队列面 * @return string */ private function changeCurrentSide(){ $currentSide = memcache_get(self::$client, $this->queueName . self::SIDE_KEY); if($currentSide == 'A'){ memcache_set(self::$client, $this->queueName . self::SIDE_KEY,'B',false,$this->expire); $this->currentSide = 'B'; }else{ memcache_set(self::$client, $this->queueName . self::SIDE_KEY,'A',false,$this->expire); $this->currentSide = 'A'; } return $this->currentSide; } /* * 检查当前队列是否已满 * @return boolean */ public function isFull(){ $result = false; if($this->sideATail == self::MAXNUM && $this->sideBTail == self::MAXNUM){ $result = true; } return $result; } /* * 检查当前队列是否为空 * @return boolean */ public function isEmpty(){ $result = true; if($this->sideATail > 0 || $this->sideBTail > 0){ $result = false; } return $result; } /* * 获取当前队列的长度 * 该长度为理论长度,某些元素由于过期失效而丢失,真实长度小于或等于该长度 * @return int */ public function getQueueLength(){ $this->getHeadNTail($this->queueName); $this->getCurrentSide(); $sideALength = $this->sideATail - $this->sideAHead; $sideBLength = $this->sideBTail - $this->sideBHead; $result = $sideALength + $sideBLength; return $result; } /* * 清空当前队列数据,仅保留HEAD_KEY、TAIL_KEY、SIDE_KEY三个key * @return boolean */ public function clear(){ if(!$this->getLock()) return false; for($i=0;$i<self::MAXNUM;$i++){ @memcache_delete(self::$client, $this->queueName.'A'. self::VALU_KEY .$i, 0); @memcache_delete(self::$client, $this->queueName.'B'. self::VALU_KEY .$i, 0); } $this->unLock(); $this->resetSide('A'); $this->resetSide('B'); return true; } /* * 清除所有memcache缓存数据 * @return NULL */ public function memFlush(){ memcache_flush(self::$client); } }