搜尋
首頁後端開發php教程PHP+memcache实现消息队列案例分享_PHP

memcache

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

php实现消息队列的用处比如在做发送邮件时发送大量邮件很费时间的问题,那么可以采取队列。
方便实现队列的轻量级队列服务器是:
starling支持memcache协议的轻量级持久化服务器
https://github.com/starling/starling
Beanstalkd轻量、高效,支持持久化,每秒可处理3000左右的队列
http://kr.github.com/beanstalkd/
php中也可以使用memcache/memcached来实现消息队列。

复制代码 代码如下:
        /**
    * Memcache 消息队列类
    */ 
    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 计数器,增加计数并返回新的计数
    * @param string $key   计数器
    * @param int $offset   计数增量,可为负数.0为不改变计数
    * @param int $time     时间
    * @return int/false    失败是返回false,成功时返回更新计数器后的计数
    */ 
    static public function set_counter( $key, $offset, $time=0 ){ 
    $mc = self::mc_init(); 
    $val = $mc->get($key); 
    if( !is_numeric($val) || $val     $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     return $mc->decrement( $key, -$offset ); 
    } 
    return $val; 
    } 
    /**
    * 写入队列
    * @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 ); 
    } 
    /**
    * 读取队列里的数据
    * @param string $key
    * @param int $max  最多读取条数
    * @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     $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 );//写入队列
    $list = QMC::output($key);//读取队列
    */ 
    ?> 

基于PHP共享内存实现的消息队列:

复制代码 代码如下:
/**
* 使用共享内存的PHP循环内存队列实现
* 支持多进程, 支持各种数据类型的存储
* 注: 完成入队或出队操作,尽快使用unset(), 以释放临界区
*
* @author wangbinandi@gmail.com
* @created 2009-12-23
*/ 
class ShmQueue 

private $maxQSize = 0; // 队列最大长度 
private $front = 0; // 队头指针 
private $rear = 0;  // 队尾指针 
private $blockSize = 256;  // 块的大小(byte) 
private $memSize = 25600;  // 最大共享内存(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; 
// 申?一个信号量 
$this->semId = sem_get($shmkey, 1); 
sem_acquire($this->semId); // 申请进入临界区 
$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 ){ // 队满 
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 ){ // 队空 
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); // 出临界区, 释放信号量 


/*
// 进队操作
$shmq = new ShmQueue();
$data = 'test data';
$shmq->enQueue($data);
unset($shmq);
// 出队操作
$shmq = new ShmQueue();
$data = $shmq->deQueue();
unset($shmq);
*/ 
?>

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

复制代码 代码如下:
    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     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;  
    }  
    }  

陳述
本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn
PHP和Python:解釋了不同的範例PHP和Python:解釋了不同的範例Apr 18, 2025 am 12:26 AM

PHP主要是過程式編程,但也支持面向對象編程(OOP);Python支持多種範式,包括OOP、函數式和過程式編程。 PHP適合web開發,Python適用於多種應用,如數據分析和機器學習。

PHP和Python:深入了解他們的歷史PHP和Python:深入了解他們的歷史Apr 18, 2025 am 12:25 AM

PHP起源於1994年,由RasmusLerdorf開發,最初用於跟踪網站訪問者,逐漸演變為服務器端腳本語言,廣泛應用於網頁開發。 Python由GuidovanRossum於1980年代末開發,1991年首次發布,強調代碼可讀性和簡潔性,適用於科學計算、數據分析等領域。

在PHP和Python之間進行選擇:指南在PHP和Python之間進行選擇:指南Apr 18, 2025 am 12:24 AM

PHP適合網頁開發和快速原型開發,Python適用於數據科學和機器學習。 1.PHP用於動態網頁開發,語法簡單,適合快速開發。 2.Python語法簡潔,適用於多領域,庫生態系統強大。

PHP和框架:現代化語言PHP和框架:現代化語言Apr 18, 2025 am 12:14 AM

PHP在現代化進程中仍然重要,因為它支持大量網站和應用,並通過框架適應開發需求。 1.PHP7提升了性能並引入了新功能。 2.現代框架如Laravel、Symfony和CodeIgniter簡化開發,提高代碼質量。 3.性能優化和最佳實踐進一步提升應用效率。

PHP的影響:網絡開發及以後PHP的影響:網絡開發及以後Apr 18, 2025 am 12:10 AM

PHPhassignificantlyimpactedwebdevelopmentandextendsbeyondit.1)ItpowersmajorplatformslikeWordPressandexcelsindatabaseinteractions.2)PHP'sadaptabilityallowsittoscaleforlargeapplicationsusingframeworkslikeLaravel.3)Beyondweb,PHPisusedincommand-linescrip

PHP類型提示如何起作用,包括標量類型,返回類型,聯合類型和無效類型?PHP類型提示如何起作用,包括標量類型,返回類型,聯合類型和無效類型?Apr 17, 2025 am 12:25 AM

PHP類型提示提升代碼質量和可讀性。 1)標量類型提示:自PHP7.0起,允許在函數參數中指定基本數據類型,如int、float等。 2)返回類型提示:確保函數返回值類型的一致性。 3)聯合類型提示:自PHP8.0起,允許在函數參數或返回值中指定多個類型。 4)可空類型提示:允許包含null值,處理可能返回空值的函數。

PHP如何處理對象克隆(克隆關鍵字)和__clone魔法方法?PHP如何處理對象克隆(克隆關鍵字)和__clone魔法方法?Apr 17, 2025 am 12:24 AM

PHP中使用clone關鍵字創建對象副本,並通過\_\_clone魔法方法定制克隆行為。 1.使用clone關鍵字進行淺拷貝,克隆對象的屬性但不克隆對象屬性內的對象。 2.通過\_\_clone方法可以深拷貝嵌套對象,避免淺拷貝問題。 3.注意避免克隆中的循環引用和性能問題,優化克隆操作以提高效率。

PHP與Python:用例和應用程序PHP與Python:用例和應用程序Apr 17, 2025 am 12:23 AM

PHP適用於Web開發和內容管理系統,Python適合數據科學、機器學習和自動化腳本。 1.PHP在構建快速、可擴展的網站和應用程序方面表現出色,常用於WordPress等CMS。 2.Python在數據科學和機器學習領域表現卓越,擁有豐富的庫如NumPy和TensorFlow。

See all articles

熱AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智慧驅動的應用程序,用於創建逼真的裸體照片

AI Clothes Remover

AI Clothes Remover

用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool

Undress AI Tool

免費脫衣圖片

Clothoff.io

Clothoff.io

AI脫衣器

AI Hentai Generator

AI Hentai Generator

免費產生 AI 無盡。

熱門文章

R.E.P.O.能量晶體解釋及其做什麼(黃色晶體)
1 個月前By尊渡假赌尊渡假赌尊渡假赌
R.E.P.O.最佳圖形設置
1 個月前By尊渡假赌尊渡假赌尊渡假赌
威爾R.E.P.O.有交叉遊戲嗎?
1 個月前By尊渡假赌尊渡假赌尊渡假赌

熱工具

MinGW - Minimalist GNU for Windows

MinGW - Minimalist GNU for Windows

這個專案正在遷移到osdn.net/projects/mingw的過程中,你可以繼續在那裡關注我們。 MinGW:GNU編譯器集合(GCC)的本機Windows移植版本,可自由分發的導入函式庫和用於建置本機Windows應用程式的頭檔;包括對MSVC執行時間的擴展,以支援C99功能。 MinGW的所有軟體都可以在64位元Windows平台上運作。

SublimeText3 英文版

SublimeText3 英文版

推薦:為Win版本,支援程式碼提示!

SublimeText3漢化版

SublimeText3漢化版

中文版,非常好用

SAP NetWeaver Server Adapter for Eclipse

SAP NetWeaver Server Adapter for Eclipse

將Eclipse與SAP NetWeaver應用伺服器整合。

PhpStorm Mac 版本

PhpStorm Mac 版本

最新(2018.2.1 )專業的PHP整合開發工具