PHP+memcache实现消息队列案例分享

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

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

代码如下:

<?php 
    /**
    * 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 < 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; 
    } 
    /**
    * 写入队列
    * @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 < 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 );//写入队列
    $list = QMC::output($key);//读取队列
    */ 
    ?>

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

代码如下:

<?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 < $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;  
    }  
    }

(0)

相关推荐

  • PHP消息队列用法实例分析

    本文实例讲述了PHP消息队列用法.分享给大家供大家参考,具体如下: 该消息队列用于linux下,进程通信 #根据路径和后缀创建一个id $key = ftok(__DIR__, 'R'); #获取队列中的消息 $q = msg_get_queue($key); #删除队列 msg_remove_queue($q); #获取队列的状态信息 $status = msg_stat_queue($q); print_r($status); echo "\n"; for($i=0;$i<1

  • php Memcache 中实现消息队列

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

  • phpredis提高消息队列的实时性方法(推荐)

    数据库存贮都用list形式 要存2个队列 1个用作消息队列保存到数据 还有个 就是用来实时读取数据在redis $redis->lpush($queenkey, json_encode($array)); $redis->lpush($listkey, json_encode($array)); /*消息队列实例*/ public function insertinfo() { $infos = array('info1' => mt_rand(10,100), 'info2' =>

  • PHP基于Redis消息队列实现发布微博的方法

    本文实例讲述了PHP基于Redis消息队列实现发布微博的方法.分享给大家供大家参考,具体如下: phpRedisAdmin :github地址  图形化管理界面 git clone [url]https://github.com/ErikDubbelboer/phpRedisAdmin.git[/url] cd phpRedisAdmin git clone [url]https://github.com/nrk/predis.git[/url] vendor 首先安装上述的Redis图形化管理

  • 队列在编程中的实际应用(php)

    一:队列的概念.数据结构 队列(Queue)是运算受到限制的一种线性表.只允许在表的一端进行插入,而在另一端进行删除元素的线性表.队尾(rear)是允许插入的一端.队头(front)是允许删除的一端.空队列是不含元素的空表. 假设有个队列Q=(a1,a2,-,an),则a1为队头元素,an为队尾元素.元素入队的次序为a1,a2,-,an,而出队的次序为a1,a2,-,an.可见队列的操作是按照先进先出的原则进行的. 其他详细的介绍请在网上搜索很多资料. 二:PHP的队列 在PHP中队列以数组的形

  • PHP使用数组实现队列

    PHP中将数组当做一个栈,主要是使用array_push()和array_pop()两个系统函数来完成.入栈主要是利用array_push()函数向第一个参数的数组尾部添加一个或多个元素,然后返回新数组的长度,示例如下: 而PHP中,将数组当作是队列则主要是用array_push()和array_shift()实现. 复制代码 代码如下: <?php $zhan=array("WEB");//声明一个数组当做队列 array_push($zhan,"PHP")

  • PHP队列用法实例

    本文实例讲述了PHP队列用法.分享给大家供大家参考.具体分析如下: 什么是队列,是先进先出的线性表,在具体应用中通常用链表或者数组来实现,队列只允许在后端进行插入操作,在前端进行删除操作. 什么情况下会用了队列呢,并发请求又要保证事务的完整性的时候就会用到队列,当然不排除使用其它更好的方法,知道的不仿说说看. 队列还可以用于减轻数据库服务器压力,我们可以将不是即时数据放入到队列中,在数据库空闲的时候或者间隔一段时间后执行.比如访问计数器,没有必要即时的执行访问增加的Sql,在没有使用队列的时候s

  • PHP实现的memcache环形队列类实例

    本文实例讲述了PHP实现的memcache环形队列类.分享给大家供大家参考.具体如下: 这里介绍了PHP实现的memcache环形队列类.没咋学过数据结构,因为业务需要,所以只是硬着头皮模拟的! 参考PHP memcache 队列代码.为使队列随时可入可出,且不受int长度越界危险(单链采取Head自增的话不作处理有越界可能),所以索性改写成环形队列.可能还有BUG,忘见谅! <?php /** * PHP memcache 环形队列类 * 原作者 LKK/lianq.net * 修改 FoxH

  • PHP队列原理及基于队列的写文件案例

    本文实例讲述了PHP队列原理及基于队列的写文件案例.分享给大家供大家参考,具体如下: 队列是一种线性表,按照先进先出的原则进行的: 入队: 出队: PHP实现队列:第一个元素作为队头,最后一个元素作为队尾 <?php /** * 队列就是这么简单 * * @link */ $array = array('PHP', 'JAVA'); array_push($array, 'PYTHON'); //入队列 array_shift($array); //出队列 什么是双端队列(或双向队列)Deque

  • php实现的双向队列类实例

    本文实例讲述了php实现的双向队列类及其用法,对于PHP数据结构与算法的学习有不错的参考价值.分享给大家供大家参考.具体分析如下: (deque,全名double-ended queue)是一种具有队列和栈的性质的数据结构.双向队列中的元素可以从两端弹出,其限定插入和删除操作在表的两端进行. 在实际使用中,还可以有输出受限的双向队列(即一个端点允许插入和删除,另一个端点只允许插入的双向队列)和输入受限的双向队列(即一个端点允许插入和删除,另一个端点只允许删除的双向队列).而如果限定双向队列从某个

  • PHP的Laravel框架中使用消息队列queue及异步队列的方法

    queue配置 首先说明一下我之前的项目中如何使用queue的. 我们现在的项目都是用的symfony,老一点的项目用的symfony1.4,新一点的项目用的都是symfony2.symfony用起来整体感觉还是很爽的,尤其symfony2,整体上来讲使用了很多java里面框架的设计思想.但是他不支持queue.在symfony,我们使用queue也经历了几个过程.最开始使用张堰同学的httpsqs.这个简单使用,但是存在单点.毕竟我们的项目还是正式对外服务的,所以我们研究了Apache旗下的开

随机推荐