搜索

首页  >  文章  >  后端开发  >  php实现的memcached队列类

php实现的memcached队列类

WBOY
WBOY原创
2016-07-25 08:43:27881浏览
  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 $value_key = $this->queueName .$this->currentSide . self::VALU_KEY . $this->currentTail;
  166. if(memcache_add(self::$client, $value_key, $data, false, $this->expire)){
  167. $this->changeTail();
  168. $result = true;
  169. }
  170. }else{//当前队列已满,更换轮值面
  171. $this->unLock();
  172. $this->changeCurrentSide();
  173. return $this->add($data);
  174. }
  176. $this->unLock();
  177. return $result;
  178. }
  180. /*
  181. * 取出数据
  182. * @param [length] int 数据的长度
  183. * @return array
  184. */
  185. public function get($length=0){
  186. if(!is_numeric($length)) return false;
  187. if(empty($length)) $length = self::MAXNUM * 2;//默认读取所有
  188. if(!$this->getLock()) return false;
  190. if($this->isEmpty()){
  191. $this->unLock();
  192. return false;
  193. }
  195. $keyArray = $this->getKeyArray($length);
  196. $lastKey = $keyArray['lastKey'];
  197. $currentKey = $keyArray['currentKey'];
  198. $keys = $keyArray['keys'];
  199. $this->changeHead($this->lastSide,$lastKey);
  200. $this->changeHead($this->currentSide,$currentKey);
  202. $data = @memcache_get(self::$client, $keys);
  203. foreach($keys as $v){//取出之后删除
  204. @memcache_delete(self::$client, $v, 0);
  205. }
  206. $this->unLock();
  208. return $data;
  209. }
  211. /*
  212. * 读取数据
  213. * @param [length] int 数据的长度
  214. * @return array
  215. */
  216. public function read($length=0){
  217. if(!is_numeric($length)) return false;
  218. if(empty($length)) $length = self::MAXNUM * 2;//默认读取所有
  219. $keyArray = $this->getKeyArray($length);
  220. $data = @memcache_get(self::$client, $keyArray['keys']);
  221. return $data;
  222. }
  224. /*
  225. * 获取队列某段长度的key数组
  226. * @param [length] int 队列长度
  227. * @return array
  228. */
  229. private function getKeyArray($length){
  230. $result = array('keys'=>array(),'lastKey'=>array(),'currentKey'=>array());
  231. $this->getHeadNTail($this->queueName);
  232. $this->getCurrentSide();
  233. if(empty($length)) return $result;
  235. //先取上一面的key
  236. $i = $result['lastKey'] = 0;
  237. for($i=0;$i $result['lastKey'] = $this->lastHead + $i;
  238. if($result['lastKey'] >= $this->lastTail) break;
  239. $result['keys'][] = $this->queueName .$this->lastSide . self::VALU_KEY . $result['lastKey'];
  240. }
  242. //再取当前面的key
  243. $j = $length - $i;
  244. $k = $result['currentKey'] = 0;
  245. for($k=0;$k $result['currentKey'] = $this->currentHead + $k;
  246. if($result['currentKey'] >= $this->currentTail) break;
  247. $result['keys'][] = $this->queueName .$this->currentSide . self::VALU_KEY . $result['currentKey'];
  248. }
  250. return $result;
  251. }
  253. /*
  254. * 更新当前轮值面队列尾的值
  255. * @return NULL
  256. */
  257. private function changeTail(){
  258. $tail_key = $this->queueName .$this->currentSide . self::TAIL_KEY;
  259. memcache_add(self::$client, $tail_key, 0,false, $this->expire);//如果没有,则插入;有则false;
  260. //memcache_increment(self::$client, $tail_key, 1);//队列尾+1
  261. $v = memcache_get(self::$client, $tail_key) +1;
  262. memcache_set(self::$client, $tail_key,$v,false,$this->expire);
  263. }
  265. /*
  266. * 更新队列首的值
  267. * @param [side] string 要更新的面
  268. * @param [headValue] int 队列首的值
  269. * @return NULL
  270. */
  271. private function changeHead($side,$headValue){
  272. if($headValue $head_key = $this->queueName .$side . self::HEAD_KEY;
  273. $tail_key = $this->queueName .$side . self::TAIL_KEY;
  274. $sideTail = memcache_get(self::$client, $tail_key);
  275. if($headValue memcache_set(self::$client, $head_key,$headValue+1,false,$this->expire);
  276. }elseif($headValue >= $sideTail){
  277. $this->resetSide($side);
  278. }
  279. }
  281. /*
  282. * 重置队列面,即将该队列面的队首、队尾值置为0
  283. * @param [side] string 要重置的面
  284. * @return NULL
  285. */
  286. private function resetSide($side){
  287. $head_key = $this->queueName .$side . self::HEAD_KEY;
  288. $tail_key = $this->queueName .$side . self::TAIL_KEY;
  289. memcache_set(self::$client, $head_key,0,false,$this->expire);
  290. memcache_set(self::$client, $tail_key,0,false,$this->expire);
  291. }
  293. /*
  294. * 改变当前轮值队列面
  295. * @return string
  296. */
  297. private function changeCurrentSide(){
  298. $currentSide = memcache_get(self::$client, $this->queueName . self::SIDE_KEY);
  299. if($currentSide == 'A'){
  300. memcache_set(self::$client, $this->queueName . self::SIDE_KEY,'B',false,$this->expire);
  301. $this->currentSide = 'B';
  302. }else{
  303. memcache_set(self::$client, $this->queueName . self::SIDE_KEY,'A',false,$this->expire);
  304. $this->currentSide = 'A';
  305. }
  306. return $this->currentSide;
  307. }
  309. /*
  310. * 检查当前队列是否已满
  311. * @return boolean
  312. */
  313. public function isFull(){
  314. $result = false;
  315. if($this->sideATail == self::MAXNUM && $this->sideBTail == self::MAXNUM){
  316. $result = true;
  317. }
  318. return $result;
  319. }
  321. /*
  322. * 检查当前队列是否为空
  323. * @return boolean
  324. */
  325. public function isEmpty(){
  326. $result = true;
  327. if($this->sideATail > 0 || $this->sideBTail > 0){
  328. $result = false;
  329. }
  330. return $result;
  331. }
  333. /*
  334. * 获取当前队列的长度
  335. * 该长度为理论长度，某些元素由于过期失效而丢失，真实长度小于或等于该长度
  336. * @return int
  337. */
  338. public function getQueueLength(){
  339. $this->getHeadNTail($this->queueName);
  340. $this->getCurrentSide();
  342. $sideALength = $this->sideATail - $this->sideAHead;
  343. $sideBLength = $this->sideBTail - $this->sideBHead;
  344. $result = $sideALength + $sideBLength;
  346. return $result;
  347. }
  349. /*
  350. * 清空当前队列数据,仅保留HEAD_KEY、TAIL_KEY、SIDE_KEY三个key
  351. * @return boolean
  352. */
  353. public function clear(){
  354. if(!$this->getLock()) return false;
  355. for($i=0;$i<:maxnum> @memcache_delete(self::$client, $this->queueName.'A'. self::VALU_KEY .$i, 0);
  356. @memcache_delete(self::$client, $this->queueName.'B'. self::VALU_KEY .$i, 0);
  357. }
  358. $this->unLock();
  359. $this->resetSide('A');
  360. $this->resetSide('B');
  361. return true;
  362. }
  364. /*
  365. * 清除所有memcache缓存数据
  366. * @return NULL
  367. */
  368. public function memFlush(){
  369. memcache_flush(self::$client);
  370. }
  372. }
复制代码

php, memcached


声明：
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系admin@php.cn
上一篇：一个简单的php分页类代码 下一篇：PHP计算地图上两点间的距离

相关文章

查看更多