当前位置：Gxlcms > PHP教程 > php实现的memcached队列类

php实现的memcached队列类

时间：2021-07-01 10:21:17 帮助过：6人阅读

  1. /*
  2. * memcache队列类
  3. * 支持多进程并发写入、读取
  4. * 边写边读,AB面轮值替换
  5. * @author guoyu
  6. * @create on 9:25 2014-9-28
  7. * @qq技术行业交流群：136112330
  8. *
  9. * @example:
  10. * $obj = new memcacheQueue('duilie');
  11. * $obj->add('1asdf');
  12. * $obj->getQueueLength();
  13. * $obj->read(11);
  14. * $obj->get(8);
  15. */
  17. class memcacheQueue{
  18. public static $client; //memcache客户端连接
  19. public $access; //队列是否可更新
  20. private $currentSide; //当前轮值的队列面:A/B
  21. private $lastSide; //上一轮值的队列面:A/B
  22. private $sideAHead; //A面队首值
  23. private $sideATail; //A面队尾值
  24. private $sideBHead; //B面队首值
  25. private $sideBTail; //B面队尾值
  26. private $currentHead; //当前队首值
  27. private $currentTail; //当前队尾值
  28. private $lastHead; //上轮队首值
  29. private $lastTail; //上轮队尾值
  30. private $expire; //过期时间,秒,1~2592000,即30天内;0为永不过期
  31. private $sleepTime; //等待解锁时间,微秒
  32. private $queueName; //队列名称,唯一值
  33. private $retryNum; //重试次数,= 10 * 理论并发数
  35. const MAXNUM = 2000; //(单面)最大队列数,建议上限10K
  36. const HEAD_KEY = '_lkkQueueHead_'; //队列首kye
  37. const TAIL_KEY = '_lkkQueueTail_'; //队列尾key
  38. const VALU_KEY = '_lkkQueueValu_'; //队列值key
  39. const LOCK_KEY = '_lkkQueueLock_'; //队列锁key
  40. const SIDE_KEY = '_lkkQueueSide_'; //轮值面key
  42. /*
  43. * 构造函数
  44. * @param [config] array memcache服务器参数
  45. * @param [queueName] string 队列名称
  46. * @param [expire] string 过期时间
  47. * @return NULL
  48. */
  49. public function __construct($queueName ='',$expire='',$config =''){
  50. if(empty($config)){
  51. self::$client = memcache_pconnect('localhost',11211);
  52. }elseif(is_array($config)){//array('host'=>'127.0.0.1','port'=>'11211')
  53. self::$client = memcache_pconnect($config['host'],$config['port']);
  54. }elseif(is_string($config)){//"127.0.0.1:11211"
  55. $tmp = explode(':',$config);
  56. $conf['host'] = isset($tmp[0]) ? $tmp[0] : '127.0.0.1';
  57. $conf['port'] = isset($tmp[1]) ? $tmp[1] : '11211';
  58. self::$client = memcache_pconnect($conf['host'],$conf['port']);
  59. }
  60. if(!self::$client) return false;
  62. ignore_user_abort(TRUE);//当客户断开连接,允许继续执行
  63. set_time_limit(0);//取消脚本执行延时上限
  65. $this->access = false;
  66. $this->sleepTime = 1000;
  67. $expire = (empty($expire) && $expire!=0) ? 3600 : (int)$expire;
  68. $this->expire = $expire;
  69. $this->queueName = $queueName;
  70. $this->retryNum = 10000;
  72. $side = memcache_add(self::$client, $queueName . self::SIDE_KEY, 'A',false, $expire);
  73. $this->getHeadNTail($queueName);
  74. if(!isset($this->sideAHead) || empty($this->sideAHead)) $this->sideAHead = 0;
  75. if(!isset($this->sideATail) || empty($this->sideATail)) $this->sideATail = 0;
  76. if(!isset($this->sideBHead) || empty($this->sideBHead)) $this->sideBHead = 0;
  77. if(!isset($this->sideBHead) || empty($this->sideBHead)) $this->sideBHead = 0;
  78. }
  80. /*
  81. * 获取队列首尾值
  82. * @param [queueName] string 队列名称
  83. * @return NULL
  84. */
  85. private function getHeadNTail($queueName){
  86. $this->sideAHead = (int)memcache_get(self::$client, $queueName.'A'. self::HEAD_KEY);
  87. $this->sideATail = (int)memcache_get(self::$client, $queueName.'A'. self::TAIL_KEY);
  88. $this->sideBHead = (int)memcache_get(self::$client, $queueName.'B'. self::HEAD_KEY);
  89. $this->sideBTail = (int)memcache_get(self::$client, $queueName.'B'. self::TAIL_KEY);
  90. }
  92. /*
  93. * 获取当前轮值的队列面
  94. * @return string 队列面名称
  95. */
  96. public function getCurrentSide(){
  97. $currentSide = memcache_get(self::$client, $this->queueName . self::SIDE_KEY);
  98. if($currentSide == 'A'){
  99. $this->currentSide = 'A';
  100. $this->lastSide = 'B';
  102. $this->currentHead = $this->sideAHead;
  103. $this->currentTail = $this->sideATail;
  104. $this->lastHead = $this->sideBHead;
  105. $this->lastTail = $this->sideBTail;
  106. }else{
  107. $this->currentSide = 'B';
  108. $this->lastSide = 'A';
  110. $this->currentHead = $this->sideBHead;
  111. $this->currentTail = $this->sideBTail;
  112. $this->lastHead = $this->sideAHead;
  113. $this->lastTail = $this->sideATail;
  114. }
  116. return $this->currentSide;
  117. }
  119. /*
  120. * 队列加锁
  121. * @return boolean
  122. */
  123. private function getLock(){
  124. if($this->access === false){
  125. while(!memcache_add(self::$client, $this->queueName .self::LOCK_KEY, 1, false, $this->expire) ){
  126. usleep($this->sleepTime);
  127. @$i++;
  128. if($i > $this->retryNum){//尝试等待N次
  129. return false;
  130. break;
  131. }
  132. }
  133. return $this->access = true;
  134. }
  135. return false;
  136. }
  138. /*
  139. * 队列解锁
  140. * @return NULL
  141. */
  142. private function unLock(){
  143. memcache_delete(self::$client, $this->queueName .self::LOCK_KEY);
  144. $this->access = false;
  145. }
  147. /*
  148. * 添加数据
  149. * @param [data] 要存储的值
  150. * @return boolean
  151. */
  152. public function add($data){
  153. $result = false;
  154. if(!$this->getLock()){
  155. return $result;
  156. }
  157. $this->getHeadNTail($this->queueName);
  158. $this->getCurrentSide();
  160. if($this->isFull()){
  161. $this->unLock();
  162. return false;
  163. }
  165. if($this->currentTail < self::MAXNUM){
  166. $value_key = $this->queueName .$this->currentSide . self::VALU_KEY . $this->currentTail;
  167. if(memcache_add(self::$client, $value_key, $data, false, $this->expire)){
  168. $this->changeTail();
  169. $result = true;
  170. }
  171. }else{//当前队列已满,更换轮值面
  172. $this->unLock();
  173. $this->changeCurrentSide();
  174. return $this->add($data);
  175. }
  177. $this->unLock();
  178. return $result;
  179. }
  181. /*
  182. * 取出数据
  183. * @param [length] int 数据的长度
  184. * @return array
  185. */
  186. public function get($length=0){
  187. if(!is_numeric($length)) return false;
  188. if(empty($length)) $length = self::MAXNUM * 2;//默认读取所有
  189. if(!$this->getLock()) return false;
  191. if($this->isEmpty()){
  192. $this->unLock();
  193. return false;
  194. }
  196. $keyArray = $this->getKeyArray($length);
  197. $lastKey = $keyArray['lastKey'];
  198. $currentKey = $keyArray['currentKey'];
  199. $keys = $keyArray['keys'];
  200. $this->changeHead($this->lastSide,$lastKey);
  201. $this->changeHead($this->currentSide,$currentKey);
  203. $data = @memcache_get(self::$client, $keys);
  204. foreach($keys as $v){//取出之后删除
  205. @memcache_delete(self::$client, $v, 0);
  206. }
  207. $this->unLock();
  209. return $data;
  210. }
  212. /*
  213. * 读取数据
  214. * @param [length] int 数据的长度
  215. * @return array
  216. */
  217. public function read($length=0){
  218. if(!is_numeric($length)) return false;
  219. if(empty($length)) $length = self::MAXNUM * 2;//默认读取所有
  220. $keyArray = $this->getKeyArray($length);
  221. $data = @memcache_get(self::$client, $keyArray['keys']);
  222. return $data;
  223. }
  225. /*
  226. * 获取队列某段长度的key数组
  227. * @param [length] int 队列长度
  228. * @return array
  229. */
  230. private function getKeyArray($length){
  231. $result = array('keys'=>array(),'lastKey'=>array(),'currentKey'=>array());
  232. $this->getHeadNTail($this->queueName);
  233. $this->getCurrentSide();
  234. if(empty($length)) return $result;
  236. //先取上一面的key
  237. $i = $result['lastKey'] = 0;
  238. for($i=0;$i<$length;$i++){
  239. $result['lastKey'] = $this->lastHead + $i;
  240. if($result['lastKey'] >= $this->lastTail) break;
  241. $result['keys'][] = $this->queueName .$this->lastSide . self::VALU_KEY . $result['lastKey'];
  242. }
  244. //再取当前面的key
  245. $j = $length - $i;
  246. $k = $result['currentKey'] = 0;
  247. for($k=0;$k<$j;$k++){
  248. $result['currentKey'] = $this->currentHead + $k;
  249. if($result['currentKey'] >= $this->currentTail) break;
  250. $result['keys'][] = $this->queueName .$this->currentSide . self::VALU_KEY . $result['currentKey'];
  251. }
  253. return $result;
  254. }
  256. /*
  257. * 更新当前轮值面队列尾的值
  258. * @return NULL
  259. */
  260. private function changeTail(){
  261. $tail_key = $this->queueName .$this->currentSide . self::TAIL_KEY;
  262. memcache_add(self::$client, $tail_key, 0,false, $this->expire);//如果没有,则插入;有则false;
  263. //memcache_increment(self::$client, $tail_key, 1);//队列尾+1
  264. $v = memcache_get(self::$client, $tail_key) +1;
  265. memcache_set(self::$client, $tail_key,$v,false,$this->expire);
  266. }
  268. /*
  269. * 更新队列首的值
  270. * @param [side] string 要更新的面
  271. * @param [headValue] int 队列首的值
  272. * @return NULL
  273. */
  274. private function changeHead($side,$headValue){
  275. if($headValue < 1) return false;
  276. $head_key = $this->queueName .$side . self::HEAD_KEY;
  277. $tail_key = $this->queueName .$side . self::TAIL_KEY;
  278. $sideTail = memcache_get(self::$client, $tail_key);
  279. if($headValue < $sideTail){
  280. memcache_set(self::$client, $head_key,$headValue+1,false,$this->expire);
  281. }elseif($headValue >= $sideTail){
  282. $this->resetSide($side);
  283. }
  284. }
  286. /*
  287. * 重置队列面,即将该队列面的队首、队尾值置为0
  288. * @param [side] string 要重置的面
  289. * @return NULL
  290. */
  291. private function resetSide($side){
  292. $head_key = $this->queueName .$side . self::HEAD_KEY;
  293. $tail_key = $this->queueName .$side . self::TAIL_KEY;
  294. memcache_set(self::$client, $head_key,0,false,$this->expire);
  295. memcache_set(self::$client, $tail_key,0,false,$this->expire);
  296. }
  298. /*
  299. * 改变当前轮值队列面
  300. * @return string
  301. */
  302. private function changeCurrentSide(){
  303. $currentSide = memcache_get(self::$client, $this->queueName . self::SIDE_KEY);
  304. if($currentSide == 'A'){
  305. memcache_set(self::$client, $this->queueName . self::SIDE_KEY,'B',false,$this->expire);
  306. $this->currentSide = 'B';
  307. }else{
  308. memcache_set(self::$client, $this->queueName . self::SIDE_KEY,'A',false,$this->expire);
  309. $this->currentSide = 'A';
  310. }
  311. return $this->currentSide;
  312. }
  314. /*
  315. * 检查当前队列是否已满
  316. * @return boolean
  317. */
  318. public function isFull(){
  319. $result = false;
  320. if($this->sideATail == self::MAXNUM && $this->sideBTail == self::MAXNUM){
  321. $result = true;
  322. }
  323. return $result;
  324. }
  326. /*
  327. * 检查当前队列是否为空
  328. * @return boolean
  329. */
  330. public function isEmpty(){
  331. $result = true;
  332. if($this->sideATail > 0 || $this->sideBTail > 0){
  333. $result = false;
  334. }
  335. return $result;
  336. }
  338. /*
  339. * 获取当前队列的长度
  340. * 该长度为理论长度，某些元素由于过期失效而丢失，真实长度小于或等于该长度
  341. * @return int
  342. */
  343. public function getQueueLength(){
  344. $this->getHeadNTail($this->queueName);
  345. $this->getCurrentSide();
  347. $sideALength = $this->sideATail - $this->sideAHead;
  348. $sideBLength = $this->sideBTail - $this->sideBHead;
  349. $result = $sideALength + $sideBLength;
  351. return $result;
  352. }
  354. /*
  355. * 清空当前队列数据,仅保留HEAD_KEY、TAIL_KEY、SIDE_KEY三个key
  356. * @return boolean
  357. */
  358. public function clear(){
  359. if(!$this->getLock()) return false;
  360. for($i=0;$i @memcache_delete(self::$client, $this->queueName.'A'. self::VALU_KEY .$i, 0);
  361. @memcache_delete(self::$client, $this->queueName.'B'. self::VALU_KEY .$i, 0);
  362. }
  363. $this->unLock();
  364. $this->resetSide('A');
  365. $this->resetSide('B');
  366. return true;
  367. }
  369. /*
  370. * 清除所有memcache缓存数据
  371. * @return NULL
  372. */
  373. public function memFlush(){
  374. memcache_flush(self::$client);
  375. }
  377. }

php, memcached

人气教程排行

本站所有资源全部来源于网络，若本站发布的内容侵害到您的隐私或者利益，请联系我们删除！