PHP memcache实现消息队列实例

memche消息队列的原理就是在key上做文章,用以做一个连续的数字加上前缀记录序列化以后消息或者日志,然后通过定时程序将内容落地到文件或者数据库.

php实现消息队列的用处比如在做发送邮件时发送大量邮件很费时间的问题,那么可以采取队列.

方便实现队列的轻量级队列服务器是:

starling支持memcache协议的轻量级持久化服务器

https://github.com/starling/starling

Beanstalkd轻量、高效，支持持久化，每秒可处理3000左右的队列

http://kr.github.com/beanstalkd/

php中也可以使用memcache/memcached来实现消息队列,代码如下:

1. <?php 
2. /** 
3. * Memcache 消息队列类 
4. */ 
5.  
6. class QMC { 
7. const PREFIX = 'ASDFASDFFWQKE'; 
8.  
9. /** 
10. * 初始化mc 
11. * @staticvar string $mc 
12. * @return Memcache 
13. */ 
14. static private function mc_init() { 
15. static $mc = null; 
16. if (is_null($mc)) { 
17. $mc = new Memcache; 
18. $mc->connect('127.0.0.1', 11211); 
19. } 
20. return $mc; 
21. } 
22. /** 
23. * mc 计数器,增加计数并返回新的计数 
24. * @param string $key   计数器 
25. * @param int $offset   计数增量,可为负数.0为不改变计数 
26. * @param int $time     时间 
27. * @return int/false    失败是返回false,成功时返回更新计数器后的计数 
28. */ 
29. static public function set_counter( $key, $offset, $time=0 ){ 
30. $mc = self::mc_init(); 
31. $val = $mc->get($key); 
32. if( !is_numeric($val) || $val < 0 ){ 
33. $ret = $mc->set( $key, 0, $time ); 
34. if( !$ret ) return false; 
35. $val = 0; 
36. } 
37. $offset = intval( $offset ); 
38. if( $offset > 0 ){ 
39. return $mc->increment( $key, $offset ); 
40. }elseif( $offset < 0 ){ 
41. return $mc->decrement( $key, -$offset ); 
42. } 
43. return $val; 
44. } 
45.  
46. /** 
47. * 写入队列 
48. * @param string $key 
49. * @param mixed $value 
50. * @return bool 
51. */ 
52. static public function input( $key, $value ){ 
53. $mc = self::mc_init(); 
54. $w_key = self::PREFIX.$key.'W'; 
55. $v_key = self::PREFIX.$key.self::set_counter($w_key, 1); 
56. return $mc->set( $v_key, $value ); 
57. } 
58. /** 
59. * 读取队列里的数据 
60. * @param string $key 
61. * @param int $max  最多读取条数 
62. * @return array 
63. */ 
64. static public function output( $key, $max=100 ){ 
65. $out = array(); 
66. $mc = self::mc_init(); 
67. $r_key = self::PREFIX.$key.'R'; 
68. $w_key = self::PREFIX.$key.'W'; 
69. $r_p   = self::set_counter( $r_key, 0 );//读指针 
70. $w_p   = self::set_counter( $w_key, 0 );//写指针 
71. if( $r_p == 0 ) $r_p = 1; 
72. while( $w_p >= $r_p ){ 
73. if( --$max < 0 ) break; 
74. $v_key = self::PREFIX.$key.$r_p; 
75. $r_p = self::set_counter( $r_key, 1 ); 
76. $out[] = $mc->get( $v_key ); 
77. $mc->delete($v_key); 
78. }//开源代码phpfensi.com 
79. return $out; 
80. } 
81. } 
82. /** 
83. 使用方法: 
84. QMC::input($key, $value );//写入队列 
85. $list = QMC::output($key);//读取队列 
86. */ 
87. ?> 

基于PHP共享内存实现的消息队列,代码如下:

1. <?php 
2. /** 
3. * 使用共享内存的PHP循环内存队列实现 
4. * 支持多进程, 支持各种数据类型的存储 
5. * 注: 完成入队或出队操作,尽快使用unset(), 以释放临界区 
6. * 
7. * @author wangbinandi@gmail.com 
8. * @created 2009-12-23 
9. */ 
10. class ShmQueue 
11. { 
12. private $maxQSize = 0; // 队列最大长度 
13.  
14. private $front = 0; // 队头指针 
15. private $rear = 0;  // 队尾指针 
16.  
17. private $blockSize = 256;  // 块的大小(byte) 
18. private $memSize = 25600;  // 最大共享内存(byte) 
19. private $shmId = 0; 
20.  
21. private $filePtr = './shmq.ptr'; 
22.  
23. private $semId = 0; 
24. public function __construct() 
25. { 
26. $shmkey = ftok(__FILE__, 't'); 
27.  
28. $this->shmId = shmop_open($shmkey, "c", 0644, $this->memSize ); 
29. $this->maxQSize = $this->memSize / $this->blockSize; 
30.  
31. //一个信号量 
32. $this->semId = sem_get($shmkey, 1); 
33. sem_acquire($this->semId); // 申请进入临界区 
34.  
35. $this->init(); 
36. } 
37.  
38. private function init() 
39. { 
40. if ( file_exists($this->filePtr) ){ 
41. $contents = file_get_contents($this->filePtr); 
42. $data = explode( '|', $contents ); 
43. if ( isset($data[0]) && isset($data[1])){ 
44. $this->front = (int)$data[0]; 
45. $this->rear  = (int)$data[1]; 
46. } 
47. } 
48. } 
49.  
50. public function getLength() 
51. { 
52. return (($this->rear - $this->front + $this->memSize) % ($this->memSize) )/$this->blockSize; 
53. } 
54.  
55. public function enQueue( $value ) 
56. { 
57. if ( $this->ptrInc($this->rear) == $this->front ){ // 队满 
58. return false; 
59. } 
60.  
61. $data = $this->encode($value); 
62. shmop_write($this->shmId, $data, $this->rear ); 
63. $this->rear = $this->ptrInc($this->rear); 
64. return true; 
65. } 
66.  
67. public function deQueue() 
68. { 
69. if ( $this->front == $this->rear ){ // 队空 
70. return false; 
71. } 
72. $value = shmop_read($this->shmId, $this->front, $this->blockSize-1); 
73. $this->front = $this->ptrInc($this->front); 
74. return $this->decode($value); 
75. } 
76.  
77. private function ptrInc( $ptr ) 
78. { 
79. return ($ptr + $this->blockSize) % ($this->memSize); 
80. } 
81.  
82. private function encode( $value ) 
83. { 
84. $data = serialize($value) . "__eof"; 
85. echo ''; 
86.  
87. echo strlen($data); 
88. echo ''; 
89.  
90. echo $this->blockSize -1; 
91. echo ''; 
92.  
93. if ( strlen($data) > $this->blockSize -1 ){ 
94. throw new Exception(strlen($data)." is overload block size!"); 
95. } 
96. return $data; 
97. } 
98.  
99. private function decode( $value ) 
100. { 
101. $data = explode("__eof", $value); 
102. return unserialize($data[0]); 
103. } 
104.  
105. public function __destruct() 
106. { 
107. $data = $this->front . '|' . $this->rear; 
108. file_put_contents($this->filePtr, $data); 
109.  
110. sem_release($this->semId); // 出临界区, 释放信号量 
111. } 
112. } 
113.  
114. /* 
115. // 进队操作 
116. $shmq = new ShmQueue(); 
117. $data = 'test data'; 
118. $shmq->enQueue($data); 
119. unset($shmq); 
120. // 出队操作 
121. $shmq = new ShmQueue(); 
122. $data = $shmq->deQueue(); 
123. unset($shmq); 
124. */ 
125. ?> 

对于一个很大的消息队列，频繁进行进行大数据库的序列化 和 反序列化，有太耗费。下面是我用PHP 实现的一个消息队列，只需要在尾部插入一个数据，就操作尾部，不用操作整个消息队列进行读取，与操作。但是，这个消息队列不是线程安全的，我只是尽量的避免了冲突的可能性。如果消息不是非常的密集，比如几秒钟才一个，还是可以考虑这样使用的。 

如果你要实现线程安全的,一个建议是通过文件进行锁定,然后进行操作,下面是代码:

1. class Memcache_Queue  
2. {  
3. private $memcache;  
4. private $name;  
5. private $prefix;  
6. function __construct($maxSize, $name, $memcache, $prefix = "__memcache_queue__")  
7. {  
8. if ($memcache == null) {  
9. throw new Exception("memcache object is null, new the object first.");  
10. }  
11. $this->memcache = $memcache;  
12. $this->name = $name;  
13. $this->prefix = $prefix;  
14. $this->maxSize = $maxSize;  
15. $this->front = 0;  
16. $this->real = 0;  
17. $this->size = 0;  
18. }  
19. function __get($name)  
20. {  
21. return $this->get($name);  
22. }  
23. function __set($name, $value)  
24. {  
25. $this->add($name, $value);  
26. return $this;  
27. }  
28. function isEmpty()  
29. {  
30. return $this->size == 0;  
31. }  
32. function isFull()  
33. {  
34. return $this->size == $this->maxSize;  
35. }  
36. function enQueue($data)  
37. {  
38. if ($this->isFull()) {  
39. throw new Exception("Queue is Full");  
40. }  
41. $this->increment("size");  
42. $this->set($this->real, $data);  
43. $this->set("real", ($this->real + 1) % $this->maxSize);  
44. return $this;  
45. }  
46. function deQueue()  
47. {  
48. if ($this->isEmpty()) {  
49. throw new Exception("Queue is Empty");  
50. }  
51. $this->decrement("size");  
52. $this->delete($this->front);  
53. $this->set("front", ($this->front + 1) % $this->maxSize);  
54. return $this;  
55. }  
56. function getTop()  
57. {  
58. return $this->get($this->front);  
59. }  
60. function getAll()  
61. {  
62. return $this->getPage();  
63. }  
64. function getPage($offset = 0, $limit = 0)  
65. {  
66. if ($this->isEmpty() || $this->size < $offset) {  
67. return null;  
68. }  
69. $keys[] = $this->getKeyByPos(($this->front + $offset) % $this->maxSize);  
70. $num = 1;  
71. for ($pos = ($this->front + $offset + 1) % $this->maxSize; $pos != $this->real; $pos = ($pos + 1) % $this->maxSize)  
72. {  
73. $keys[] = $this->getKeyByPos($pos);  
74. $num++;  
75. if ($limit > 0 && $limit == $num) {  
76. break;  
77. }  
78. }  
79. return array_values($this->memcache->get($keys));  
80. }  
81. function makeEmpty()  
82. {  
83. $keys = $this->getAllKeys();  
84. foreach ($keys as $value) {  
85. $this->delete($value);  
86. }  
87. $this->delete("real");  
88. $this->delete("front");  
89. $this->delete("size");  
90. $this->delete("maxSize");  
91. }  
92. private function getAllKeys()  
93. {  
94. if ($this->isEmpty())  
95. {  
96. return array();  
97. }  
98. $keys[] = $this->getKeyByPos($this->front);  
99. for ($pos = ($this->front + 1) % $this->maxSize; $pos != $this->real; $pos = ($pos + 1) % $this->maxSize)  
100. {  
101. $keys[] = $this->getKeyByPos($pos);  
102. }  
103. return $keys;  
104. }  
105. private function add($pos, $data)  
106. {  
107. $this->memcache->add($this->getKeyByPos($pos), $data);  
108. return $this;  
109. }  
110. private function increment($pos)  
111. {  
112. return $this->memcache->increment($this->getKeyByPos($pos));  
113. }  
114. private function decrement($pos)  
115. {  
116. $this->memcache->decrement($this->getKeyByPos($pos));  
117. }  
118. private function set($pos, $data)  
119. {  
120. $this->memcache->set($this->getKeyByPos($pos), $data);  
121. return $this;  
122. }  
123. private function get($pos)  
124. {  
125. return $this->memcache->get($this->getKeyByPos($pos));  
126. }  
127. private function delete($pos)  
128. {  
129. return $this->memcache->delete($this->getKeyByPos($pos));  
130. }  
131. private function getKeyByPos($pos)  
132. {  
133. return $this->prefix . $this->name . $pos;  
134. }  
135. }