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. }//開源代碼Vevb.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. }