Beanstalkd工作队列

Beanstalkd 是什么

Beanstalkd是目前一个绝对可靠,易于安装的消息传递服务&＃xff0c;主要用例是管理不同部分和工人之间的工作流应用程序的部署通过工作队列和消息堆栈,类似于其他受欢迎的解决方案,比如RabbitMQ。然而,创建Beanstalkd使它有别于其他工作。Beanstalkd旨在成为一个工作队列,而不是一把雨伞工具来满足许多需求。为了实现这一目的,它作为一种轻量级的、快速有效的应用程序基于C编程语言。精益建筑还允许它是安装和使用非常简单,使它适合大多数用例。

Beanstalkd 的 Features(特性)

• 持久性&＃xff1a;Beanstalkd运行使用内存,但也提供了持久性支持。
• 优先级&＃xff1a;与大多数选择一样,Beanstalkd提供了不同的任务的优先级来处理紧急事情时需要。
• 分布 &＃xff1a; 不同的服务器实例可以分布类似于Memcached是如何工作的。
• 掩盖 &＃xff1a;有可能通过掩盖它无限期延期的作业(即任务)。
• 第三方工具&＃xff1a;Beanstalkd附带各种第三方工具包括综合领先指标和基于web的管理控制台。
• .过期 &＃xff1a;工作可以设置为过期,auto-queue之后(TTR – Time To Run)

Beanstalkd使用案例

• 允许web服务器快速响应请求,而不是被迫当场曾推高程序执行
• 在指定的时间间隔执行某些工作(即爬行web)
• 分发到多个工作人员进行处理
• 让离线客户端(例如一个断开连接的用户)获取数据在稍后的时间,而不是让它永久失去
• 极大地提高应用程序的可靠性和正常运行时间

Beanstalkd 的安装

################# 第一种安装方法&＃xff0c;但是要安装git
yum install -y git
git clone https://github.com/kr/beanstalkd
cd beanstalkd
make
make install
################### 第二种源码安装
wget http://cloud.github.com/downloads/kr/beanstalkd/beanstalkd-1.4.6.tar.gz
tar xzf beanstalkd-1.4.6.tar.gz
cd beanstalkd-1.4.6
./configure
make
make instal
############# 查看安装的版本
beanstalkd -v

Beanstalkd 的使用

• 使用说明

# Usage: beanstalkd -l [ip address] -p [port #]
beanstalkd -l 127.0.0.1 -p 11301 &
# ip可以使用这台机器的ip&＃xff0c;端口可以更改&＃xff1b;
#如果想切换到后台运行&＃xff1a;
nohup beanstalkd -l 127.0.0.1 -p 11301 & #----------------------------如果不适用源文件安装的启动
# To start the service:
service beanstalkd start
# To stop the service:
service beanstalkd stop
# To restart the service:
service beanstalkd restart
# To check the status:
service beanstalkd status

python 调用 beanstalkd

安装beanstalkc

pip install pyyaml
pip install beanstalkc

基本操作

• 连接

import beanstalkc
beanstalk &＃61; beanstalkc.Connection(host&＃61;&＃39;localhost&＃39;, port&＃61;11301)

• put&＃xff08;建立一个job&＃xff09;

将一个job加入到当前tube中、然后马上设置job状态为“Ready”比如
beanstalk.put(&＃39;hey!&＃39;)
# 注意put指令只能接受字符串、任何非字符串如整型45都是错误的、迩还可以推迟它的执行、让她的状态延迟N秒后才把状态设置为“Ready”、如
beanstalk.put(&＃39;yes!&＃39;, delay&＃61;1)
# 还可以为该job设置一个优先级、最高优先级是0、然后数字越大优先级越低、
beanstalk.put(&＃39;foo&＃39;, priority&＃61;42)
beanstalk.put(&＃39;bar&＃39;, priority&＃61;21)
beanstalk.put(&＃39;qux&＃39;, priority&＃61;21)

• release&＃xff08;提前释放一个delay的job到Ready队列&＃xff09;

如果你使用了delay来延迟一个job、后来又反悔了、那么你可以使用release来提前释放这个job、让它早日回归当前tube并设置为“Ready”状态、但迩不可以对非当前tube的job使用release命令、要获取非当前tube的job可以通过peek命令beanstalk.put(&＃39;yes!&＃39;, delay&＃61;900)job.release()job.stats()[&＃39;state&＃39;]
##################### 输出
&＃39;ready&＃39;

• reserve(获取一个job)

put是往tube里加入job、那么reserve指令则是从tube里取出job、默认情况下、reserve是按job的序号从小取到大、如果像上面使用put设置了job的优先级那么是按优先级高的job取值、如果优先级同等级、那么再按序号从小取到大
beanstalk.reserve(timeout&＃61;0) is None
########################## 输出
True
-----------------------------------------------------------------------------
另外reserve可以如上所示设置timeout的时间参数、如果把时间设置为0、则任何时候都直接报timeout而造成获取job失败、迩可以设置任何一个非负数作为timeout的时间、如果超过timeout的时间、那么这次的获取job的动作就会结束、等待下次再次调用reserve指令来获取job、而没有被获取的job则将置于“Ready”状态等待下次再次被reserve所调用

• delete(删除一个job)

任何时间job都可以调用delete指令来删除这个job
job.delete()

• bury(让一个job睡觉去)

和使用delay参数不同、如果你不使用kick把它踢醒、它将永远处于沉睡状态、但迩不可以对非当前tube的job使用bury命令、要获取非当前tube的job可以通过peek命令
job &＃61; beanstalk.reserve()
job.bury()
job.stats()[&＃39;state&＃39;]
######################### 输出
&＃39;buried&＃39;

• kick(把一个正在睡觉的job叫醒)

默认情况下、先叫醒最早睡觉的那个job
beanstalk.kick()
#这个命令还可以加个参数设置唤醒的数量、比如加上42、一次过唤醒42个job、还按照最先睡觉的那个job开始、先就是先进先出原则
beanstalk.kick(42)
###################### 输出
1
0

• peek&＃xff08;查找并切换成目标job&＃xff09;

#查找序号为3的job、也就是第三个建立的job、如果该序号不存在、那么返回空值None
job &＃61; beanstalk.peek(3)
job.body
###################### 输出
&＃39;yes!&＃39;

• peek_ready (查找并切换当前tube最先进入准备状态的job)

#使用beanstalk获取最早一个处于“Ready”状态的job、注意、只能获取当前tube的job
job &＃61; beanstalk.peek_ready()
job.body
####################### 输出
&＃39;yes!&＃39;

• peek_delayed(查找并切换当前tube延迟时间最短的job)

beanstalk.put(&＃39;o tempores&＃39;, delay&＃61;120)
job &＃61; beanstalk.peek_delayed()
job.stats()[&＃39;state&＃39;]
######################## 输出
4
&＃39;delayed&＃39;

• peek_buried&＃xff08;查找并切换当前tube最先睡觉的job&＃xff09;

beanstalk.put(&＃39;o mores!&＃39;)
job &＃61; beanstalk.reserve()
job.bury()
job &＃61; beanstalk.peek_buried()
job.stats()[&＃39;state&＃39;]
##############################输出
5
&＃39;buried&＃39;

• tube&＃xff08;队列&＃xff09;的操作

#tubes或watching&＃xff08;查看所有tube&＃xff09;&＃xff1a;查看所有的tube
beanstalk.tubes()
[&＃39;default&＃39;]
#默认情况下、只存在一个tube、而这个tube的名称叫作“default”、这个管道一定会存在、不能被删除、其它的tube你可以任意删除
#ignore(删除tube)&＃xff1a;你可以使用ignore命令来删除任意非“default”的管道
beanstalk.ignore(&＃39;weibo&＃39;)
################输出
1
# use或watch(添加或者切换一个tube)&＃xff1a;你可以使用use或watch来添加或者切换一个tube、如果tube名称不存在、则创建一个同名tube、如果存在就切换到该tube下
beanstalk.use(&＃39;foo&＃39;)
##############输出
&＃39;foo&＃39;
#using(查看当前使用的tube) &＃xff1a;你可以使用using来查看当前使用的tube是什么
beanstalk.using()
################输出
&＃39;default&＃39;

• 关闭beanstalk连接

eanstalk.close()

• 样例

import beanstalkc
import json
import sys
bean &＃61; beanstalkc.Connection(&＃39;172.31.25.12&＃39;,11300)
#print bean.tubes()
#bean.use(&＃39;test&＃39;) # 添加一个队列bean.watch(&＃39;test&＃39;) # 切换到改队列
#bean.put(&＃39;task1&＃39;)
job &＃61; bean.reserve()while job:print job.bodyjob.delete()job &＃61; bean.reserve(timeout &＃61; 60)if job is None:sys.exit(0)# 如果不适用delete 删除数据&＃xff0c;就会使得放入reserved队列中&＃xff0c;过一段时间就会重新返回到ready队列。
# 当所有队列中

beanstalkc的文档