php 使用beanstalk 消息队列

Beanstalkd 消息队列

一.基本信息
Beanstalkd，一个高性能、轻量级的分布式内存队列系统，最初设计的目的是想经过后台异步执行耗时的任务来下降高容量Web应用系统的页面访问延迟，支持过有9.5 million用户的Facebook Causes应用。后来开源，如今有PostRank大规模部署和使用，天天处理百万级任务。Beanstalkd是典型的类Memcached设计，协议和使用方式都是一样的风格，因此使用过memcached的用户会以为Beanstalkd似曾相识。

 

二 特性

Beanstalkd 是一个轻量级消息中间件，它最大特色是将本身定位为基于管道  (tube) 和任务 (job) 的工做队列 (work-queue)：

Beanstalkd 支持任务优先级 (priority), 延时 (delay), 超时重发 (time-to-run) 和预留 (buried), 可以很好的支持分布式的后台任务和定时任务处理。

它的内部实现采用 libevent, 服务器-客户端之间用相似 memcached 的轻量级通信协议，具备有很高的性能。

尽管是内存队列, beanstalkd 提供了 binlog 机制, 当重启 beanstalkd 时，当前任务状态可以从纪录的本地 binlog 中恢复。

1）管道（tube）

管道相似于消息主题 (topic), 在一个 Beanstalkd 中能够支持多个管道, 每一个管道都有本身的发布者 (producer) 和消费者 (consumer). 管道之间互相不影响。

2）任务（job）

Beanstalkd 用任务 (job) 代替消息 (message) 的概念。与消息不一样，任务有一系列状态：

READY- 须要当即处理的任务，当延时 (DELAYED) 任务到期后会自动成为当前任务；
DELAYED- 延迟执行的任务, 当消费者处理任务后, 能够用将消息再次放回 DELAYED 队列延迟执行；
RESERVED- 已经被消费者获取, 正在执行的任务。Beanstalkd 负责检查任务是否在 TTR(time-to-run) 内完成；
BURIED- 保留的任务: 任务不会被执行，也不会消失，除非有人把它 "踢" 回队列；
DELETED- 消息被完全删除。Beanstalkd 再也不维持这些消息。

3）任务优先级 (priority)

任务 (job) 能够有 0~2^32 个优先级, 0 表明最高优先级。 beanstalkd 采用最大最小堆 (Min-max heap) 处理任务优先级排序, 任什么时候刻调用 reserve 命令的消费者老是能拿到当前优先级最高的任务, 时间复杂度为 O(logn).

4）延时任务 (delay)

有两种方式能够延时执行任务 (job): 生产者发布任务时指定延时；或者当任务处理完毕后, 消费者再次将任务放入队列延时执行 (RELEASE with <delay>)。

5）任务超时重发 (time-to-run)

Beanstalkd 把任务返回给消费者之后：消费者必须在预设的 TTR (time-to-run) 时间内发送 delete / release/ bury 改变任务状态；不然 Beanstalkd 会认为消息处理失败，而后把任务交给另外的消费者节点执行。若是消费者预计在 TTR (time-to-run) 时间内没法完成任务, 也能够发送 touch 命令, 它的做用是让 Beanstalkd 从系统时间从新计算 TTR (time-to-run).

6）任务预留 (buried)

若是任务由于某些缘由没法执行, 消费者能够把任务置为 buried 状态让 Beanstalkd 保留这些任务。管理员能够经过 peek buried 命令查询被保留的任务，而且进行人工干预。简单的, kick <n> 可以一次性把 n 条被保留的任务踢回队列。

 

二 安装

请看安装教程：http://www.cnblogs.com/sien6/p/8177023.html

 

三 Beanstalkd 协议

Beanstalkd 采用类 memcached 协议, 客户端经过文本命令与服务器交互。

这些命令能够简单的分红三组:

1）生产类 - use <tube> / put <priority> <delay> <ttr> [bytes]:  

生产者用 use 选择一个管道 (tube), 而后用 put 命令向管道发布任务 (job).    

2）消费类  - watch <tubes> / reserve / delete <id> / release <id> <priority> <delay> / bury <id> / touch <id>

消费者用 watch 选择多个管道 (tube), 而后用 reserve 命令获取待执行的任务，这个命令是阻塞的。客户端直到有任务可执行才返回。当任务处理完毕后, 消费者能够完全删除任务 (DELETE), 释听任务让别人处理 (RELEASE), 或者保留 (BURY) 任务。

3） 维护类 - peek job / peek delayed / peek ready / peek buried / kick <n>

用于维护管道内的任务状态, 在不改变任务状态的条件下获取任务。能够用消费类命令改变这些任务的状态。

被保留 (buried) 的任务能够用 kick 命令 "踢" 回队列。

 

应用：

生成者：

require_once 'beanstalk/Client.php';
$config = include 'beanstalk/config.php';
$beanstalk = new Beanstalk\Client($config);
$beanstalk->connect();
$beanstalk->useTube('u_tube'); //使用u_tabe 管道
$param = $argv;
$beanstalk->put(0,0,10,json_encode($param));

 

消费者：

require_once 'beanstalk/Client.php';
$config = include 'beanstalk/config.php';
$Beanstalk= new Beanstalk\Client($config);
$Beanstalk->connect();
$Beanstalk->useTube('u_tube');
$Beanstalk->watch('u_tube');    // 监控u_tube 通道
$Beanstalk->ignore('default');  // 取消默认通道监控

while(true){    // 持续监控
    $job = $Beanstalk->reserve();   // 取出待执行任务，并执行
//    print_r($Beanstalk->statsTube('foo'));
    if($job){
        $Beanstalk->bury($job['id'],'u_tube');  // 任务加入保留队列
    }
//    $Beanstalk->kick(3);  // 开启保留队列
    $Beanstalk->delete($job['id']);     //删除吧任务
}

 

参考：http://blog.csdn.net/black_ox/article/details/24792489

https://www.cnblogs.com/fuland/p/4245386.html