php實作的memcached佇列類

PHP、記憶體快取

  /*
 * memcache佇列類別
 * 支援多重行程並發寫入、讀取
 * 邊寫邊讀,AB面輪值替換
 * @author guoyu
 * @create on 9:25 2014-9-28
 * @qq科技業交流群：136112330
 *
 * @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
 constnLU_KEY = 'lkkQuue > const LOCK_KEY = '_lkkQueueLock_'; //佇列鎖定key
 const SIDE_KEY = '_lkkQueueSide_'; //輪值面key

 /*
 * 建構子

 **
 * 建構子

 *param ] array memcache伺服器參數
 * @param [queueName] string 佇列名稱
 * @param [expire] string 過期時間
 * @return NULL
 */
 public function __construct($queue ='',$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($queueue) {
 $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->;存取=== false){
 while(!memcache_add(self::$client, $this->queueName .self::LOCK_KEY, 1, false, $this->expire) ){
 usleep ($這個->睡眠時間）；
@$i;
 if($i > $this->retryNum){//嘗試等待N次
 return false;
 休息；
 }
 }
 return $this->access = true;
 }
 回傳 false;
 }

 /*
 * 佇列解鎖
 * @return NULL
 */
 private function unLock({🎜> private function unLock({ 🎜> memcache_delete(self::$client, $this->queueName .self::LOCK_KEY);
 $this->access = false;
 }

 /*
 * 新增資料
 * @param [data] 要儲存的值
 * @return boolean
 */
 public function add( $data){
 $結果= false;
 if(!$this->getLock()){
 return $result;
 }
 $this->getHeadNTail($this->queueName);
 $this->getCurrentSide();

 if($this->isFull()){
 $this->unLock();
 回傳 false;
 }

 if($this->currentTail  $value_key = $this->queueName .$this->currentSide . self::VALU_KEY 。
 if(memcache_add(self::$client, $value_key, $data, false, $this->expire)){
 $this->changeTail();
 $結果 = true;
 }
 }else{//目前佇列已滿,換輪值面
 $this->unLock();
 $this->changeCurrentSide();
 return $this->add($data);
 }

 $this->unLock();
 回傳$結果；
 }

 /*
 * 資料擷取
 * @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();
 回傳 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,self::$client, $v, 0);
 }
 $this->unLock();

 回傳$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']);
 傳回$資料;
 }

 /*
 * 取得佇列某段長度的密實取得佇列某段長度的密實取得佇列某段長度的密實取得佇列某段長度的密實取得佇列某段長度鑰數組
 * @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 $結果;

 // 先取上邊的key
 $i = $result['lastKey'] = 0;
 for($i =0;$i $result['lastKey'] = $this->lastHead $i;
 if($result['lastKey'] >= $this-> lastTail) 中斷;
 $result['keys'][] = $this->queueName .$this->lastSide 。 self::VALU_KEY 。 $結果['lastKey'];
 }

 // 再取當前面的key
 $j = $length - $i;
 $k = $result['currentKey' ] = 0;
 for($k=0;$k $result['currentKey'] = $this->currentHead $k;
 if($result ['currentKey'] >= $this->currentTail) 中斷;
 $result['keys'][] = $this->queueName .$this->currentSide 。 self::VALU_KEY 。 $結果['當前Key'];
 }

 回傳$結果；
 }

 /*
 *更新目前輪值面隊列尾的值
 * @return NULL
 */
 private function changeTail(){
 $tail_key = $this->queueName .$this->currentSide 。自我::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  $head_key = $this->queueName .$side 。自我::HEAD_KEY;
 $tail_key = $this->queueName .$side 。自我::TAIL_KEY;
 $sideTail = memcache_get(self::$client, $tail_key);
 if($headValue  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 .自我::HEAD_KEY;
 $tail_key = $this->queueName .$side 。自我::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(){
 $currentSget客戶端， $this->queueName 。 ,'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 ;
 ifif ($this->sideATail == self::MAXNUM && $this->sideBTail == self::MAXNUM){
 $result = true;
 }
 回傳$結果；
 }

 /*
 * 檢查目前佇列是否為空
 * @return boolean
 */
 public function isEmpty(){
 $result = true ;
 true ; ($this->sideATail > 0 || $this->sideBTail > 0){
 $result = false;
 }
 回傳$結果；
 }

 /*
 * 取得目前佇列的長度
 * 長度為理論長度，某些元素因過渡失效而遺失，真實長度小於或等於該長度
 * @return int
 */
 public function getQueueLength(){
 $this->getHeadNTail($this->queueName);
 $this->getCurrentSide();

 $sideALength = $this->ATail - $this- >sideAHead;
 $sideBLength = $this->sideBTail - $this->sideBHead;
 $結果= $sideALength $sideBLength;

 回傳$結果；
 }

 /*
 * 清空目前佇列數據，只保留HEAD_KEY、TAIL_KEY、SIDE_KEY三個鍵
 * @return boolean
 */
 public function clear( ){
 if(! $this->getLock()) return false;
 for($i=0;$i<:maxnum> @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');
回傳true;
 }

 /*
 * 清除所有memcache伺服器資料
 * @return NULL
 */
 public function memFlush(){
 memcache_flush(self:: $客戶）；
 }

  }