php之任务队列

陆砚码

2026-07-21 • 👁 0人浏览过

http://blog.s135.com/httpsqs/ 例如：邮件队列、任务队列、消息队列、Feed队列用户注册成功，而不是直接给用户发送email，而是把新注册的用户的email插入到邮件任务队列中。

由服务器给用户发送邮件。

发送成功或发送N次失败的将从队列中清除。

------------------------------- 如果是普通的网站，如平均每秒产生的队列消息不超过上百，则用低成本数据库+轮询程序，完全可行．具体思路如下： 1.　当产生消息时，使用MySQL的表保存产生的新队列消息，并记录一些运行参数 2.　服务器上开启cron(Linux),或开启自动刷新的浏览器窗口（Windows）,比如按半分钟运行一次轮询程序 3.　这些轮询程序每次取出一些队列消息，进行处理，处理成功或是处理失败N次后，删除MySQL队列表中的相关记录如此反复，即可实现．我觉得这个方式的优点是： 1.　实现最为容易，可行性最好 2.　低成本的开发或是平台部署成本在我们的生产环境中，这种方式得到了大量实践，表现良好．当然，如果确实是＂高性能＂，那么这个问题更不问题，既然需要高性能，说明网站的规模不一般，既然规模不一般，还会缺钱吗？

　不缺钱，就可以找这方面的专家去解决． -------------------------------

/**

* 使用共享内存的PHP循环内存队列实现
* 支持多进程, 支持各种数据类型的存储,需要pcntl扩展的支持。

* 注: 完成入队或出队操作,尽快使用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";
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下用Memcache 实现消息队列 Memcache 一般用于缓存服务。

但是很多时候，比如一个消息广播系统，需要一个消息队列。

直接从数据库取消息，负载往往不行。

如果将整个消息队列用一个key缓存到memcache里面，对于一个很大的消息队列，频繁进行进行大数据库的序列化和反序列化，有太耗费。

下面是我用 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 ;}} 转自：http://zhengdl126.iteye.com/blog/816575 参考：http://www.yuansir-web.com/2012/09/22/php-%E5%85%B1%E4%BA%AB%E5%86%85%E5%AD%98%E4%BB%A5%E5%8F%8A%E5%88%A9%E7%94%A8%E5%85%B1%E4%BA%AB%E5%86%85%E5%AD%98%E5%AE%9E%E7%8E%B0%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97/ http://www.laruence.com/2008/04/21/101.html

相关文章