欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识！
您现在的位置是： 首页  >  后端开发

php实现的memcached队列类

程序员文章站 2023-12-28 11:36:16
...
  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
相关标签： php实现的memcached队列类

上一篇：

下一篇：

推荐阅读