memche消息隊(duì)列的原理就是在key上做文章，用以做一個(gè)連續(xù)的數(shù)字加上前綴記錄序列化以后消息或者日志。然后通過(guò)定時(shí)程序?qū)?nèi)容落地到文件或者數(shù)據(jù)庫(kù)。

php實(shí)現(xiàn)消息隊(duì)列的用處比如在做發(fā)送郵件時(shí)發(fā)送大量郵件很費(fèi)時(shí)間的問題，那么可以采取隊(duì)列。
方便實(shí)現(xiàn)隊(duì)列的輕量級(jí)隊(duì)列服務(wù)器是：
starling支持memcache協(xié)議的輕量級(jí)持久化服務(wù)器
https://github.com/starling/starling
Beanstalkd輕量、高效，支持持久化，每秒可處理3000左右的隊(duì)列
http://kr.github.com/beanstalkd/
php中也可以使用memcache/memcached來(lái)實(shí)現(xiàn)消息隊(duì)列。

復(fù)制代碼 代碼如下:
    <?php 
    /**
    * Memcache 消息隊(duì)列類
    */ 
    class QMC { 
    const PREFIX = 'ASDFASDFFWQKE'; 
    /**
    * 初始化mc
    * @staticvar string $mc
    * @return Memcache
    */ 
    static private function mc_init() { 
    static $mc = null; 
    if (is_null($mc)) { 
    $mc = new Memcache; 
    $mc->connect('127.0.0.1', 11211); 
    } 
    return $mc; 
    } 
    /**
    * mc 計(jì)數(shù)器,增加計(jì)數(shù)并返回新的計(jì)數(shù)
    * @param string $key   計(jì)數(shù)器
    * @param int $offset   計(jì)數(shù)增量,可為負(fù)數(shù).0為不改變計(jì)數(shù)
    * @param int $time     時(shí)間
    * @return int/false    失敗是返回false,成功時(shí)返回更新計(jì)數(shù)器后的計(jì)數(shù)
    */ 
    static public function set_counter( $key, $offset, $time=0 ){ 
    $mc = self::mc_init(); 
    $val = $mc->get($key); 
    if( !is_numeric($val) || $val < 0 ){ 
    $ret = $mc->set( $key, 0, $time ); 
    if( !$ret ) return false; 
    $val = 0; 
    } 
    $offset = intval( $offset ); 
    if( $offset > 0 ){ 
    return $mc->increment( $key, $offset ); 
    }elseif( $offset < 0 ){ 
    return $mc->decrement( $key, -$offset ); 
    } 
    return $val; 
    } 
    /**
    * 寫入隊(duì)列
    * @param string $key
    * @param mixed $value
    * @return bool
    */ 
    static public function input( $key, $value ){ 
    $mc = self::mc_init(); 
    $w_key = self::PREFIX.$key.'W'; 
    $v_key = self::PREFIX.$key.self::set_counter($w_key, 1); 
    return $mc->set( $v_key, $value ); 
    } 
    /**
    * 讀取隊(duì)列里的數(shù)據(jù)
    * @param string $key
    * @param int $max  最多讀取條數(shù)
    * @return array
    */ 
    static public function output( $key, $max=100 ){ 
    $out = array(); 
    $mc = self::mc_init(); 
    $r_key = self::PREFIX.$key.'R'; 
    $w_key = self::PREFIX.$key.'W'; 
    $r_p   = self::set_counter( $r_key, 0 );//讀指針 
    $w_p   = self::set_counter( $w_key, 0 );//寫指針 
    if( $r_p == 0 ) $r_p = 1; 
    while( $w_p >= $r_p ){ 
    if( --$max < 0 ) break; 
    $v_key = self::PREFIX.$key.$r_p; 
    $r_p = self::set_counter( $r_key, 1 ); 
    $out[] = $mc->get( $v_key ); 
    $mc->delete($v_key); 
    } 
    return $out; 
    } 
    } 
    /**
    使用方法:
    QMC::input($key, $value );//寫入隊(duì)列
    $list = QMC::output($key);//讀取隊(duì)列
    */ 
    ?> 

基于php共享內(nèi)存實(shí)現(xiàn)的消息隊(duì)列:

復(fù)制代碼 代碼如下:
<?php 
/**
* 使用共享內(nèi)存的php循環(huán)內(nèi)存隊(duì)列實(shí)現(xiàn)
* 支持多進(jìn)程, 支持各種數(shù)據(jù)類型的存儲(chǔ)
* 注: 完成入隊(duì)或出隊(duì)操作,盡快使用unset(), 以釋放臨界區(qū)
*
* @author wangbinandi@gmail.com
* @created 2009-12-23
*/ 
class ShmQueue 
{ 
private $maxQSize = 0; // 隊(duì)列最大長(zhǎng)度 
private $front = 0; // 隊(duì)頭指針 
private $rear = 0;  // 隊(duì)尾指針 
private $blockSize = 256;  // 塊的大小(byte) 
private $memSize = 25600;  // 最大共享內(nèi)存(byte) 
private $shmId = 0; 
private $filePtr = './shmq.ptr'; 
private $semId = 0; 
public function __construct() 
{ 
$shmkey = ftok(__FILE__, 't'); 
$this->shmId = shmop_open($shmkey, "c", 0644, $this->memSize ); 
$this->maxQSize = $this->memSize / $this->blockSize; 
// 申?一個(gè)信號(hào)量 
$this->semId = sem_get($shmkey, 1); 
sem_acquire($this->semId); // 申請(qǐng)進(jìn)入臨界區(qū) 
$this->init(); 
} 
private function init() 
{ 
if ( file_exists($this->filePtr) ){ 
$contents = file_get_contents($this->filePtr); 
$data = explode( '|', $contents ); 
if ( isset($data[0]) && isset($data[1])){ 
$this->front = (int)$data[0]; 
$this->rear  = (int)$data[1]; 
} 
} 
} 
public function getLength() 
{ 
return (($this->rear - $this->front + $this->memSize) % ($this->memSize) )/$this->blockSize; 
} 
public function enQueue( $value ) 
{ 
if ( $this->ptrInc($this->rear) == $this->front ){ // 隊(duì)滿 
return false; 
} 
$data = $this->encode($value); 
shmop_write($this->shmId, $data, $this->rear ); 
$this->rear = $this->ptrInc($this->rear); 
return true; 
} 
public function deQueue() 
{ 
if ( $this->front == $this->rear ){ // 隊(duì)空 
return false; 
} 
$value = shmop_read($this->shmId, $this->front, $this->blockSize-1); 
$this->front = $this->ptrInc($this->front); 
return $this->decode($value); 
} 
private function ptrInc( $ptr ) 
{ 
return ($ptr + $this->blockSize) % ($this->memSize); 
} 
private function encode( $value ) 
{ 
$data = serialize($value) . "__eof"; 
echo ''; 
echo strlen($data); 
echo ''; 
echo $this->blockSize -1; 
echo ''; 
if ( strlen($data) > $this->blockSize -1 ){ 
throw new Exception(strlen($data)." is overload block size!"); 
} 
return $data; 
} 
private function decode( $value ) 
{ 
$data = explode("__eof", $value); 
return unserialize($data[0]); 
} 
public function __destruct() 
{ 
$data = $this->front . '|' . $this->rear; 
file_put_contents($this->filePtr, $data); 
sem_release($this->semId); // 出臨界區(qū), 釋放信號(hào)量 
} 
} 
/*
// 進(jìn)隊(duì)操作
$shmq = new ShmQueue();
$data = 'test data';
$shmq->enQueue($data);
unset($shmq);
// 出隊(duì)操作
$shmq = new ShmQueue();
$data = $shmq->deQueue();
unset($shmq);
*/ 
?>

對(duì)于一個(gè)很大的消息隊(duì)列，頻繁進(jìn)行進(jìn)行大數(shù)據(jù)庫(kù)的序列化 和 反序列化，有太耗費(fèi)。下面是我用php 實(shí)現(xiàn)的一個(gè)消息隊(duì)列，只需要在尾部插入一個(gè)數(shù)據(jù)，就操作尾部，不用操作整個(gè)消息隊(duì)列進(jìn)行讀取，與操作。但是，這個(gè)消息隊(duì)列不是線程安全的，我只是盡量的避免了沖突的可能性。如果消息不是非常的密集，比如幾秒鐘才一個(gè)，還是可以考慮這樣使用的。
如果你要實(shí)現(xiàn)線程安全的，一個(gè)建議是通過(guò)文件進(jìn)行鎖定，然后進(jìn)行操作。下面是代碼：
代碼如下:

復(fù)制代碼 代碼如下:
    class Memcache_Queue  
    {  
    private $memcache;  
    private $name;  
    private $prefix;  
    function __construct($maxSize, $name, $memcache, $prefix = "__memcache_queue__")  
    {  
    if ($memcache == null) {  
    throw new Exception("memcache object is null, new the object first.");  
    }  
    $this->memcache = $memcache;  
    $this->name = $name;  
    $this->prefix = $prefix;  
    $this->maxSize = $maxSize;  
    $this->front = 0;  
    $this->real = 0;  
    $this->size = 0;  
    }  
    function __get($name)  
    {  
    return $this->get($name);  
    }  
    function __set($name, $value)  
    {  
    $this->add($name, $value);  
    return $this;  
    }  
    function isEmpty()  
    {  
    return $this->size == 0;  
    }  
    function isFull()  
    {  
    return $this->size == $this->maxSize;  
    }  
    function enQueue($data)  
    {  
    if ($this->isFull()) {  
    throw new Exception("Queue is Full");  
    }  
    $this->increment("size");  
    $this->set($this->real, $data);  
    $this->set("real", ($this->real + 1) % $this->maxSize);  
    return $this;  
    }  
    function deQueue()  
    {  
    if ($this->isEmpty()) {  
    throw new Exception("Queue is Empty");  
    }  
    $this->decrement("size");  
    $this->delete($this->front);  
    $this->set("front", ($this->front + 1) % $this->maxSize);  
    return $this;  
    }  
    function getTop()  
    {  
    return $this->get($this->front);  
    }  
    function getAll()  
    {  
    return $this->getPage();  
    }  
    function getPage($offset = 0, $limit = 0)  
    {  
    if ($this->isEmpty() || $this->size < $offset) {  
    return null;  
    }  
    $keys[] = $this->getKeyByPos(($this->front + $offset) % $this->maxSize);  
    $num = 1;  
    for ($pos = ($this->front + $offset + 1) % $this->maxSize; $pos != $this->real; $pos = ($pos + 1) % $this->maxSize)  
    {  
    $keys[] = $this->getKeyByPos($pos);  
    $num++;  
    if ($limit > 0 && $limit == $num) {  
    break;  
    }  
    }  
    return array_values($this->memcache->get($keys));  
    }  
    function makeEmpty()  
    {  
    $keys = $this->getAllKeys();  
    foreach ($keys as $value) {  
    $this->delete($value);  
    }  
    $this->delete("real");  
    $this->delete("front");  
    $this->delete("size");  
    $this->delete("maxSize");  
    }  
    private function getAllKeys()  
    {  
    if ($this->isEmpty())  
    {  
    return array();  
    }  
    $keys[] = $this->getKeyByPos($this->front);  
    for ($pos = ($this->front + 1) % $this->maxSize; $pos != $this->real; $pos = ($pos + 1) % $this->maxSize)  
    {  
    $keys[] = $this->getKeyByPos($pos);  
    }  
    return $keys;  
    }  
    private function add($pos, $data)  
    {  
    $this->memcache->add($this->getKeyByPos($pos), $data);  
    return $this;  
    }  
    private function increment($pos)  
    {  
    return $this->memcache->increment($this->getKeyByPos($pos));  
    }  
    private function decrement($pos)  
    {  
    $this->memcache->decrement($this->getKeyByPos($pos));  
    }  
    private function set($pos, $data)  
    {  
    $this->memcache->set($this->getKeyByPos($pos), $data);  
    return $this;  
    }  
    private function get($pos)  
    {  
    return $this->memcache->get($this->getKeyByPos($pos));  
    }  
    private function delete($pos)  
    {  
    return $this->memcache->delete($this->getKeyByPos($pos));  
    }  
    private function getKeyByPos($pos)  
    {  
    return $this->prefix . $this->name . $pos;  
    }  
    }  

php技術(shù)：PHP+memcache實(shí)現(xiàn)消息隊(duì)列案例分享，轉(zhuǎn)載需保留來(lái)源！

鄭重聲明：本文版權(quán)歸原作者所有，轉(zhuǎn)載文章僅為傳播更多信息之目的，如作者信息標(biāo)記有誤，請(qǐng)第一時(shí)間聯(lián)系我們修改或刪除，多謝。