必威体育Betway必威体育官网
当前位置:首页 > IT技术

beanstalkd消息队列在生产环境的应用

时间:2019-08-27 11:40:00来源:IT技术作者:seo实验室小编阅读:77次「手机版」
 

hkcmd

Beanstalkd 是一个高性能的消息队列中间件,本博文宅鸟将介绍一下这个东东的使用。

一、先通过概念让大家了解Beanstalkd的特性和工作场景。

Beanstalkd 是一个轻量级消息中间件,它最大特点是将自己定位为基于管道  (tube) 和任务 (job) 的工作队列 (work-queue):

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

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

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

管道 (tube):

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

任务 (job):

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

READY- 需要立即处理的任务,当延时 (DELAYED) 任务到期后会自动成为当前任务;

DELAYED- 延迟执行的任务, 当消费者处理任务后, 可以用将消息再次放回 DELAYED 队列延迟执行;

RESERVED- 已经被消费者获取, 正在执行的任务。Beanstalkd 负责检查任务是否在 TTR(time-to-run) 内完成;

BURIED- 保留的任务: 任务不会被执行,也不会消失,除非有人把它 "踢" 回队列;

DELETED- 消息被彻底删除。Beanstalkd 不再维持这些消息。

任务优先级 (priority):

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

延时任务 (delay):

有两种方式可以延时执行任务 (job): 生产者发布任务时指定延时;或者当任务处理完毕后, 消费者再次将任务放入队列延时执行 (RElease with <delay>)。这种机制可以实现分布式的 java.util.Timer,这种分布式定时任务的优势是:如果某个消费者节点故障,任务超时重发 (time-to-run) 能够保证任务转移到另外的节点执行。

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

Beanstalkd 把任务返回给消费者以后:消费者必须在预设的 TTR (time-to-run) 时间内发送 delete / release/ bury 改变任务状态;否则 Beanstalkd 会认为消息处理失败,然后把任务交给另外的消费者节点执行。如果消费者预计在 TTR (time-to-run) 时间内无法完成任务, 也可以发送 touch 命令, 它的作用是让 Beanstalkd 从系统时间重新计算 TTR (time-to-run).

任务预留 (buried):

如果任务因为某些原因无法执行, 消费者可以把任务置为 buried 状态让 Beanstalkd 保留这些任务。管理员可以通过 peek buried 命令查询被保留的任务,并且进行人工干预。简单的, kick <n> 能够一次性把 n 条被保留的任务踢回队列。

Beanstalkd 协议:

Beanstalkd 采用类 memcached 协议, 客户端通过文本命令与服务器交互。这些命令可以简单的分成三组:    

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

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

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

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

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

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

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

  协议文档: https://raw.github.com/kr/beanstalkd/master/doc/protocol.txt

Beanstalkd 不足:

Beanstalkd 没有提供主备同步 + 故障切换机制, 在应用中有成为单点的风险。实际应用中,可以用数据库为任务 (job) 提供持久化存储。

Beanstalkd

另外, 和 memcached 类似, Beanstalkd 依赖 libevent 的单线程事件分发机制, 不能有效利用多核 cpu 的性能。这一点可以通过单机部署多个实例克服。

二、部署安装:

Beanstalkd 的安装非常简单:

ubuntu和debian下使用下面命令:

1

sudo apt-get install beanstalkd

安装后编辑配置文件:

1

vim /etc/default/beanstalkd

163040445.jpg

把START=NO改为:START=yes即可

更多关于安装可以参考官网

通过命令可以启动、停止Beanstalk

1

2

3

/etc/init.d/beanstalkd start

lsof -i:11300

/etc/init.d/beanstalkd stop

163604656.jpg

启动后,就可以通过客户端进行调用了:

Beanstalk支持多种客户端语言:

php,java,perl,c,c++,lua,Python,go,ruby等等(了解更多可以来官网)。

我们将通过php给大家介绍在生产环境下面的使用。

就拿录视频程序使用到的Beanstalk来给大家介绍:

先介绍一下程序结构:

视频录制程序分为两个方面,一个是产生录制任务的脚本(生产者),还有一个处理录制任务脚本(消费者)。

首先把php的客户端下载后,加入到项目中。下面把代码贴出来:

生产者:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

#!/usr/bin/php

<?php

require_once 'configuration.php';

require_once 'Record.class.php';

require_once 'BeanStalk.class.php';

$now=time();

$model new RecordModel ();

$records=$model->checkStartRecord($now);

//print_r($records);

//exit();

$beanstalk = BeanStalk::open ( array (

            'servers' => array (

                Configuration::$record_config['beanStak']

                ),

            'select' => 'random peek'

            ) );

$beanstalk->use_tube ( 'records' );

foreach $records as $record ) {

    $beanstalk->put ( 0, 0, 10, json_encode$record ) );

}

?>

消费者:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

<?php

require_once('config.php');

require_once('func.php');

require_once('BeanStalk.class.php');

$beanstalk = BeanStalk::open(array(

            'servers'       => array$config['beanStak'] ),

            'select'        => 'random peek'

            ));

$beanstalk->watch('records');

while(true){

    //$beanstalk->watch('records');

    $job $beanstalk->reserve_with_timeout();

    if(is_object($job)){

        $data=$job->get();

        $json=json_decode($data,true);

        print_r($json);

        if(!empty($json["live_name"])&&!empty($json["start_time"])&&!empty($json["end_time"])&&!empty($json["vod_id"])){

            //print_r($json);

            if(!empty($json["afterplay"])&&$json["afterplay"]==1)

            $cmd="{$config['afterplaycmd']} {$json["live_name"]} {$json["vod_id"]} {$json['start_time']} {$json['end_time']}";

            else

            $cmd="{$config['recordcmd']} {$json["live_name"]} {$json["vod_id"]} {$json['start_time']} {$json['end_time']}";

                

            echo $cmd;

            $chkcmd="ps -ef |grep '".$cmd."'  |grep -v 'grep'|wc -l";

            //$chkcmd="ps -ef |wc -l";

            //echo $chkcmd;

            $count=system($chkcmd);

            //echo $count;

            if($count==0)

            {

                //system($cmd);

                exec($cmd,$res,$rc);

                //print_r($res);

                //print_r($rc);

            }

            Beanstalk::delete($job);   // Delete the job.

            $info=array();

            $info["vod_id"]=$json['vod_id'];

            $info["record_msg"]="startjob";

            $data=array();

            $data["type"]="reciveRecords";

            $data["message"]=$info;

            $url=$config['recordStatus'];

            $httpcode = 200;

            $result = test_api($httpcode,$url,"post",json_encode($data));

            print_r($data);

        }

        //$beanstalk->watch('records');

    }

    sleep(1);

}

?>

下面我们介绍一个可以管理Beanstalk的php工具,地址如下

https://github.com/jimbojsb/bstools

把该工具安装后,就可以查看Beanstalk的各种情况了

165839569.jpg

165651192.jpg

另附一个 beanstalkd管理工具 

https://github.com/kr/beanstalkd/wiki/Tools

https://github.com/ptrofimov/beanstalk_console

注意这个需要开启PHP短标签支持

php.ini文件里short_open_tag = On

配置lib/include.php中servers

$config = array( 'servers' => array('192.168.0.3:11300','192.168.0.5:11300')) ;

相关阅读

一个用消息队列 的人,不知道为啥用 MQ,这就有点尴尬

消息队列为什么写这篇文章?博主有两位朋友分别是小A和小B:小A,工作于传统软件行业(某社保局的软件外包公司),每天工作内容就是和产

分享到:

栏目导航

推荐阅读

热门阅读