<?php
/**
 * PHP memcache 環(huán)形隊(duì)列類
 * 原作者 LKK/lianq.net
 * 修改 FoxHunter
 * 因業(yè)務(wù)需要只保留的隊(duì)列中的Pop和Push,修改過期時(shí)間為0即永久
 */
class MQueue
{
 public static $client;
 private $expire; //過期時(shí)間,秒,1~2592000,即30天內(nèi)
 private $sleepTime; //等待解鎖時(shí)間,微秒
 private $queueName; //隊(duì)列名稱,唯一值
 private $retryNum; //嘗試次數(shù)
 private $MAXNUM; //最大隊(duì)列容量
 private $canRewrite; //是否可以覆寫開關(guān)，滿出來的內(nèi)容從頭部開始覆蓋重寫原來的數(shù)據(jù)
 private $HEAD; //下一步要進(jìn)入的指針位置
 private $TAIL; //下一步要進(jìn)入的指針位置
 private $LEN; //隊(duì)列現(xiàn)有長(zhǎng)度
 const LOCK_KEY = '_Fox_MQ_LOCK_'; //鎖存儲(chǔ)標(biāo)示
 const LENGTH_KEY = '_Fox_MQ_LENGTH_'; //隊(duì)列現(xiàn)長(zhǎng)度存儲(chǔ)標(biāo)示
 const VALU_KEY = '_Fox_MQ_VAL_'; //隊(duì)列鍵值存儲(chǔ)標(biāo)示
 const HEAD_KEY = '_Fox_MQ_HEAD_'; //隊(duì)列HEAD指針位置標(biāo)示
 const TAIL_KEY = '_Fox_MQ_TAIL_'; //隊(duì)列TAIL指針位置標(biāo)示
 /*
  * 構(gòu)造函數(shù)
  * 對(duì)于同一個(gè)$queueName，實(shí)例化時(shí)必須保障構(gòu)造函數(shù)的參數(shù)值一致，否則pop和push會(huì)導(dǎo)隊(duì)列順序混亂
  */
 public function __construct($queueName = '', $maxqueue = 1, $canRewrite = false, $expire = 0, $config = '')
 {
  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); //當(dāng)客戶斷開連接,允許繼續(xù)執(zhí)行
  set_time_limit(0); //取消腳本執(zhí)行延時(shí)上限
  $this->access  = false;
  $this->sleepTime = 1000;
  $expire   = (empty($expire)) ? 0 : (int) $expire + 1;
  $this->expire  = $expire;
  $this->queueName = $queueName;
  $this->retryNum = 20000;
  $this->MAXNUM  = $maxqueue != null ? $maxqueue : 1;
  $this->canRewrite = $canRewrite;
  $this->getHeadAndTail();
  if (!isset($this->HEAD) || empty($this->HEAD))
   $this->HEAD = 0;
  if (!isset($this->TAIL) || empty($this->TAIL))
   $this->TAIL = 0;
  if (!isset($this->LEN) || empty($this->LEN))
   $this->LEN = 0;
 }
 //獲取隊(duì)列首尾指針信息和長(zhǎng)度
 private function getHeadAndTail()
 {
  $this->HEAD = (int) memcache_get(self::$client, $this->queueName . self::HEAD_KEY);
  $this->TAIL = (int) memcache_get(self::$client, $this->queueName . self::TAIL_KEY);
  $this->LEN = (int) memcache_get(self::$client, $this->queueName . self::LENGTH_KEY);
 }
 // 利用memcache_add原子性加鎖
 private function lock()
 {
  if ($this->access === false) {
   $i = 0;
   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;
 }
 //更新頭部指針指向,指向下一個(gè)位置
 private function incrHead()
 {
  //$this->getHeadAndTail(); //獲取最新指針信息 ,由于本方法體均在鎖內(nèi)調(diào)用，其鎖內(nèi)已調(diào)用了此方法，本行注釋
  $this->HEAD++; //頭部指針下移
  if ($this->HEAD >= $this->MAXNUM) {
   $this->HEAD = 0; //邊界值修正
  }
  ;
  $this->LEN--; //Head的移動(dòng)由Pop觸發(fā)，所以相當(dāng)于數(shù)量減少
  if ($this->LEN < 0) {
   $this->LEN = 0; //邊界值修正
  }
  ;
  memcache_set(self::$client, $this->queueName . self::HEAD_KEY, $this->HEAD, false, $this->expire); //更新
  memcache_set(self::$client, $this->queueName . self::LENGTH_KEY, $this->LEN, false, $this->expire); //更新
 }
 //更新尾部指針指向，指向下一個(gè)位置
 private function incrTail()
 {
  //$this->getHeadAndTail(); //獲取最新指針信息，由于本方法體均在鎖內(nèi)調(diào)用，其鎖內(nèi)已調(diào)用了此方法，本行注釋
  $this->TAIL++; //尾部指針下移
  if ($this->TAIL >= $this->MAXNUM) {
   $this->TAIL = 0; //邊界值修正
  }
  ;
  $this->LEN++; //Head的移動(dòng)由Push觸發(fā)，所以相當(dāng)于數(shù)量增加
  if ($this->LEN >= $this->MAXNUM) {
   $this->LEN = $this->MAXNUM; //邊界值長(zhǎng)度修正
  }
  ;
  memcache_set(self::$client, $this->queueName . self::TAIL_KEY, $this->TAIL, false, $this->expire); //更新
  memcache_set(self::$client, $this->queueName . self::LENGTH_KEY, $this->LEN, false, $this->expire); //更新
 }
 // 解鎖
 private function unLock()
 {
  memcache_delete(self::$client, $this->queueName . self::LOCK_KEY);
  $this->access = false;
 }
 //判斷是否滿隊(duì)列
 public function isFull()
 {
  //外部直接調(diào)用的時(shí)候由于沒有鎖所以此處的值是個(gè)大概值，并不很準(zhǔn)確，但是內(nèi)部調(diào)用由于在前面有l(wèi)ock，所以可信
  if ($this->canRewrite)
   return false;
  return $this->LEN == $this->MAXNUM ? true : false;
 }
 //判斷是否為空
 public function isEmpty()
 {
  //外部直接調(diào)用的時(shí)候由于沒有鎖所以此處的值是個(gè)大概值，并不很準(zhǔn)確，但是內(nèi)部調(diào)用由于在前面有l(wèi)ock，所以可信
  return $this->LEN == 0 ? true : false;
 }
 public function getLen()
 {
  //外部直接調(diào)用的時(shí)候由于沒有鎖所以此處的值是個(gè)大概值，并不很準(zhǔn)確，但是內(nèi)部調(diào)用由于在前面有l(wèi)ock，所以可信
  return $this->LEN;
 }
 /*
  * push值
  * @param mixed 值
  * @return bool
  */
 public function push($data = '')
 {
  $result = false;
  if (empty($data))
   return $result;
  if (!$this->lock()) {
   return $result;
  }
  $this->getHeadAndTail(); //獲取最新指針信息
  if ($this->isFull()) { //只有在非覆寫下才有Full概念
   $this->unLock();
   return false;
  }
  if (memcache_set(self::$client, $this->queueName . self::VALU_KEY . $this->TAIL, $data, MEMCACHE_COMPRESSED, $this->expire)) {
   //當(dāng)推送后，發(fā)現(xiàn)尾部和頭部重合（此時(shí)指針還未移動(dòng)），且右邊仍有未由Head讀取的數(shù)據(jù)，那么移動(dòng)Head指針，避免尾部指針跨越Head
   if ($this->TAIL == $this->HEAD && $this->LEN >= 1) {
    $this->incrHead();
   }
   $this->incrTail(); //移動(dòng)尾部指針
   $result = true;
  }
  $this->unLock();
  return $result;
 }
 /*
  * Pop一個(gè)值
  * @param [length] int 隊(duì)列長(zhǎng)度
  * @return array
  */
 public function pop($length = 0)
 {
  if (!is_numeric($length))
   return false;
  if (!$this->lock())
   return false;
  $this->getHeadAndTail();
  if (empty($length))
   $length = $this->LEN; //默認(rèn)讀取所有
  if ($this->isEmpty()) {
   $this->unLock();
   return false;
  }
  //獲取長(zhǎng)度超出隊(duì)列長(zhǎng)度后進(jìn)行修正
  if ($length > $this->LEN)
   $length = $this->LEN;
  $data = $this->popKeyArray($length);
  $this->unLock();
  return $data;
 }
 /*
  * pop某段長(zhǎng)度的值
  * @param [length] int 隊(duì)列長(zhǎng)度
  * @return array
  */
 private function popKeyArray($length)
 {
  $result = array();
  if (empty($length))
   return $result;
  for ($k = 0; $k < $length; $k++) {
   $result[] = @memcache_get(self::$client, $this->queueName . self::VALU_KEY . $this->HEAD);
   @memcache_delete(self::$client, $this->queueName . self::VALU_KEY . $this->HEAD, 0);
   //當(dāng)提取值后，發(fā)現(xiàn)頭部和尾部重合（此時(shí)指針還未移動(dòng)），且右邊沒有數(shù)據(jù)，即隊(duì)列中最后一個(gè)數(shù)據(jù)被完全掏空，此時(shí)指針停留在本地不移動(dòng)，隊(duì)列長(zhǎng)度變?yōu)?
   if ($this->TAIL == $this->HEAD && $this->LEN <= 1) {
    $this->LEN = 0;
    memcache_set(self::$client, $this->queueName . self::LENGTH_KEY, $this->LEN, false, $this->expire); //更新
    break;
   } else {
    $this->incrHead(); //首尾未重合，或者重合但是仍有未讀取出的數(shù)據(jù)，均移動(dòng)HEAD指針到下一處待讀取位置
   }
  }
  return $result;
 }
 /*
  * 重置隊(duì)列
  * * @return NULL
  */
 private function reset($all = false)
 {
  if ($all) {
   memcache_delete(self::$client, $this->queueName . self::HEAD_KEY, 0);
   memcache_delete(self::$client, $this->queueName . self::TAIL_KEY, 0);
   memcache_delete(self::$client, $this->queueName . self::LENGTH_KEY, 0);
  } else {
   $this->HEAD = $this->TAIL = $this->LEN = 0;
   memcache_set(self::$client, $this->queueName . self::HEAD_KEY, 0, false, $this->expire);
   memcache_set(self::$client, $this->queueName . self::TAIL_KEY, 0, false, $this->expire);
   memcache_set(self::$client, $this->queueName . self::LENGTH_KEY, 0, false, $this->expire);
  }
 }
 /*
  * 清除所有memcache緩存數(shù)據(jù)
  * @return NULL
  */
 public function memFlush()
 {
  memcache_flush(self::$client);
 }
 public function clear($all = false)
 {
  if (!$this->lock())
   return false;
  $this->getHeadAndTail();
  $Head = $this->HEAD;
  $Length = $this->LEN;
  $curr = 0;
  for ($i = 0; $i < $Length; $i++) {
   $curr = $this->$Head + $i;
   if ($curr >= $this->MAXNUM) {
    $this->HEAD = $curr = 0;
   }
   @memcache_delete(self::$client, $this->queueName . self::VALU_KEY . $curr, 0);
  }
  $this->unLock();
  $this->reset($all);
  return true;
 }
}