PHP memcache class sharing (memcache queue)_PHP tutorial

memcacheQueue.class.php

Copy code The code is as follows:

/**
 * PHP memcache 队列类
 * @author LKK/lianq.net
 * @version 0.3
 * @修改说明:
 * 1.放弃了之前的AB面轮值思路,使用类似数组的构造,重写了此类.
 * 2.队列默认先进先出,但增加了反向读取功能.
 * 3.感谢网友FoxHunter提出的宝贵意见.
 * @example:
 * $obj = new memcacheQueue('duilie');
 * $obj->add('1asdf');
 * $obj->getQueueLength();
 * $obj->read(10);
 * $obj->get(8);
 */
class memcacheQueue{
 public static $client;   //memcache客户端连接
 public   $access;   //队列是否可更新
 private   $expire;   //过期时间,秒,1~2592000,即30天内
 private   $sleepTime;   //等待解锁时间,微秒
 private   $queueName;   //队列名称,唯一值
 private   $retryNum;   //重试次数,= 10 * 理论并发数
 public   $currentHead;  //当前队首值
 public   $currentTail;  //当前队尾值

 const MAXNUM  = 20000;    //最大队列数,建议上限10K
 const HEAD_KEY = '_lkkQueueHead_';  //队列首kye
 const TAIL_KEY = '_lkkQueueTail_';  //队列尾key
 const VALU_KEY = '_lkkQueueValu_';  //队列值key
 const LOCK_KEY = '_lkkQueueLock_';  //队列锁key

/**
* Constructor
* @param string $queueName Queue name
* @param int $expire Expiration time
* @param array $config memcache configuration
*
* @return < ;type>
*/
public function __construct($queueName ='',$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);//When the client disconnects ,Allow execution to continue
set_time_limit(0);//Cancel the upper limit of script execution delay

$this->access = false;
$this->sleepTime = 1000;
$expire = empty($expire) ? 3600 : intval($expire)+1;
$this->expire = $expire;
$this->queueName = $queueName;
$this ->retryNum = 1000;

$this->head_key = $this->queueName . self::HEAD_KEY;
$this->tail_key = $this->queueName . self ::TAIL_KEY;
$this->lock_key = $this->queueName . self::LOCK_KEY;

$this->_initSetHeadNTail();
}

/**
* Initialize the queue head and tail values ​​
*/
private function _initSetHeadNTail(){
//The value of the current queue head
$this->currentHead = memcache_get(self::$client, $this-> ;head_key);
if($this->currentHead === false) $this->currentHead =0;

//The value at the end of the current queue
$this-> currentTail = memcache_get(self::$client, $this->tail_key);
if($this->currentTail === false) $this->currentTail =0;
}

/**
* When taking out an element, change the value at the head of the queue
* @param int $step step value
*/
private function _changeHead($step=1){
$this->currentHead += $step;
memcache_set(self::$client, $ this->head_key,$this->currentHead,false,$this->expire);
}

/**
* When adding elements, change the value at the end of the queue
* @param int $step step value
* @param bool $reverse Whether to reverse
* @return null
*/
private function _changeTail($step =1, $reverse =false){
if(!$reverse){
$this->currentTail += $step;
}else{
$this->currentTail -= $step;
}

memcache_set(self::$client, $this->tail_key,$this->currentTail,false,$this->expire);
}

/**
* Whether the queue is empty
* @return bool
*/
private function _isEmpty(){
return (bool)($this->currentHead === $this->currentTail);
}

/**
* Whether the queue is full
* @return bool
*/
private function _isFull(){
$len = $this->currentTail - $this->currentHead;
return (bool )($len === self::MAXNUM);
}

/**
* Queue lock
*/
private function _getLock(){
if($this-> access === false){
while(!memcache_add(self::$client, $this->lock_key, 1, false, $this->expire) ){
usleep($this-> ;sleepTime);
@$i++;
if($i > $this->retryNum){//Try to wait N times
return false;
break;
}
}

   $this->_initSetHeadNTail();
   return $this->access = true;
  }

  return $this->access;
 }

    /**
* Queue Unlocked
*/
 private function _unLock(){
  memcache_delete(self::$client, $this->lock_key, 0);
  $this->access = false;
 }

    /**
* Get the length of the current queue
* This length is the theoretical length, some elements are lost due to expiration, the real length
* @return int
*/
 public function getQueueLength(){
  $this->_initSetHeadNTail();
  return intval($this->currentTail - $this->currentHead);
 }

    /**
* Add queue data
* @param void $data The data to be added
* @return bool
*/
 public function add($data){
  if(!$this->_getLock()) return false;

  if($this->_isFull()){
   $this->_unLock();
   return false;
  }

  $value_key = $this->queueName . self::VALU_KEY . strval($this->currentTail +1);
  $result = memcache_set(self::$client, $value_key, $data, MEMCACHE_COMPRESSED, $this->expire);
  if($result){
   $this->_changeTail();
  }

  $this->_unLock();
  return $result;
 }

    /**
* Read queue data
* @param int $length The length to be read (use negative numbers for reverse reading)
* @return array
*/
 public function read($length=0){
  if(!is_numeric($length)) return false;
  $this->_initSetHeadNTail();

  if($this->_isEmpty()){
   return false;
  }

  if(empty($length)) $length = self::MAXNUM;//默认所有
  $keyArr = array();
  if($length >0){//正向读取(从队列首向队列尾)
   $tmpMin = $this->currentHead;
   $tmpMax = $tmpMin + $length;
   for($i= $tmpMin; $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 = @memcache_get(self::$client, $keyArr);

  return $result;
 }

    /**
* Take out the queue data
* @param int $length The length to be taken out (use negative numbers for reverse reading)
* @return array
*/
 public function get($length=0){
  if(!is_numeric($length)) return false;
  if(!$this->_getLock()) return false;

  if($this->_isEmpty()){
   $this->_unLock();
   return false;
  }

  if(empty($length)) $length = self::MAXNUM;//默认所有
  $length = intval($length);
  $keyArr = array();
  if($length >0){//正向读取(从队列首向队列尾)
   $tmpMin = $this->currentHead;
   $tmpMax = $tmpMin + $length;
   for($i= $tmpMin; $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 = @memcache_get(self::$client, $keyArr);

  foreach($keyArr as $v){//取出之后删除
   @memcache_delete(self::$client, $v, 0);
  }

  $this->_unLock();

  return $result;
 }

    /**
* Clear the queue
*/
 public function clear(){
  if(!$this->_getLock()) return false;

  if($this->_isEmpty()){
   $this->_unLock();
   return false;
  }

  $tmpMin = $this->currentHead--;
  $tmpMax = $this->currentTail++;

  for($i= $tmpMin; $i   $tmpKey = $this->queueName . self::VALU_KEY . $i;
   @memcache_delete(self::$client, $tmpKey, 0);
  }

  $this->currentTail = $this->currentHead = 0;
  memcache_set(self::$client, $this->head_key,$this->currentHead,false,$this->expire);
  memcache_set(self::$client, $this->tail_key,$this->currentTail,false,$this->expire);

  $this->_unLock();
 }

 /*
  * 清除所有memcache缓存数据
  */
 public function memFlush(){
  memcache_flush(self::$client);
 }

}//end class

http://www.bkjia.com/PHPjc/746623.htmlwww.bkjia.comtruehttp://www.bkjia.com/PHPjc/746623.htmlTechArticlememcacheQueue.class.php 复制代码 代码如下: ?php /** * PHP memcache 队列类 * @author LKK/lianq.net * @version 0.3 * @修改说明: * 1.放弃了之前的AB面轮值思路...