PHP操作Beanstalkd队列(1)安装与基础

释放双眼,带上耳机,听听看~!

Beanstalkd 是一款高性能、轻量级的分布式内存消息队列。
轻量是通过对比而来,相比RabbitMQ和Kafka,Beanstalkd显得更加简单易用,同时可以满足小系统的应用,可以实现生产者和消费者分离模式。

一、概念

  • job:一个需要异步处理的任务,是 Beanstalkd 中的基本单元,需要放在一个 tube 中

  • tube:一个有名的任务队列,用来存储统一类型的 job,是 producer 和 consumer 操作的对象

  • producer:Job 的生产者,通过 put 命令来将一个 job 放到一个 tube 中

  • consumer:Job的消费者,通过 reserve/release/bury/delete 命令来获取 job 或改变 job 的状态

二、生命周期

PHP操作Beanstalkd队列(1)安装与基础

producer(生产者)put一个新的job到tube时,job就处于READY状态,等待consumer来处理,如果选择延迟put,job就先到DELAYED状态,等待时间过后才迁移到READY状态。

consumer获取了当前READY的job后,该job的状态就迁移到RESERVED,这样其他的consumer就不能再操作该job。

当consumer完成该job后,可以选择delete,

release或者bury操作;delete之后,job从系统消亡,之后不能再获取;release操作可以重新把该job状态迁移回READY(也可以延迟该状态迁移操作),使其他的consumer可以继续获取和执行该job;有意思的是bury操作,可以把该job休眠,等到需要的时候,再将休眠的job
kick回READY状态,也可以delete BURIED状态的job。

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

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

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

  • BURIED – 保留的任务: 任务不会被执行,也不会消失,除非有人把它 “踢” 回队列

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

三、特性

1. 优先级(priority)

对于很紧急的消费者可以让他在队列中提前,实现插队行为

2. 延迟(delay)

设置的任务在多少秒后才能被消费者读取,可以实现定时任务,如定时点赞等

3. 持久化(persistent data)

出现服务器宕机等情况,数据依然存在

4. 预留(buried)

消费者将无法取出任务,在其他合适的时机在拿出消费

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

消费者必须在指定的时间内处理完任务,否则这个任务将重新被放入队列

四、安装

#下载软件-编译安装git clone git://github.com/kr/beanstalkd.gitcd beanstalkd#编译make#安装完成-默认安装在/usr/local/bin/#yum安装yum -y install beanstalkd --enablerepo=epel#配置文件/etc/sysconfig/beanstalkd#启动并绑定端口 beanstalkd -l 127.0.0.1 -p 11300 -b /home/software/binstalkd/binlogs#-b开启日志备份,重启恢复数据#安装php操作类composer require pda/pheanstalk

创建demo.php文件

<?php
    require './vendor/autoload.php'; 
    $p = new /Pheanstalk/Pheanstalk('127.0.0.1',11300); 
    print_r($p->stats());// 打印出队列所有信息

队列参数

   'current-jobs-urgent' => '0', // 优先级小于1024状态为ready的job数量
   'current-jobs-ready' => '0', // 状态为ready的job数量
   'current-jobs-reserved' => '0', // 状态为reserved的job数量
   'current-jobs-delayed' => '0', // 状态为delayed的job数量
   'current-jobs-buried' => '0', // 状态为buried的job数量
   'cmd-put' => '0', // 总共执行put指令的次数
   'cmd-peek' => '0', // 总共执行peek指令的次数
   'cmd-peek-ready' => '0', // 总共执行peek-ready指令的次数
   'cmd-peek-delayed' => '0', // 总共执行peek-delayed指令的次数
   'cmd-peek-buried' => '0', // 总共执行peek-buried指令的次数
   'cmd-reserve' => '0', // 总共执行reserve指令的次数
   'cmd-reserve-with-timeout' => '0',   'cmd-delete' => '0',   'cmd-release' => '0',   'cmd-use' => '0', // 总共执行use指令的次数
   'cmd-watch' => '0', // 总共执行watch指令的次数
   'cmd-ignore' => '0',   'cmd-bury' => '0',   'cmd-kick' => '0',   'cmd-touch' => '0',   'cmd-stats' => '2',   'cmd-stats-job' => '0',   'cmd-stats-tube' => '0',   'cmd-list-tubes' => '0',   'cmd-list-tube-used' => '0',   'cmd-list-tubes-watched' => '0',   'cmd-pause-tube' => '0',   'job-timeouts' => '0', // 所有超时的job的总共数量
   'total-jobs' => '0', // 创建的所有job数量
   'max-job-size' => '65535', // job的数据部分最大长度
   'current-tubes' => '1', // 当前存在的tube数量
   'current-connections' => '1', // 当前打开的连接数
   'current-producers' => '0', // 当前所有的打开的连接中至少执行一次put指令的连接数量
   'current-workers' => '0', // 当前所有的打开的连接中至少执行一次reserve指令的连接数量
   'current-waiting' => '0', // 当前所有的打开的连接中执行reserve指令但是未响应的连接数量
   'total-connections' => '2', // 总共处理的连接数
   'pid' => '3609', // 服务器进程的id
   'version' => '1.10', // 服务器版本号
   'rusage-utime' => '0.000000', // 进程总共占用的用户CPU时间
   'rusage-stime' => '0.001478', // 进程总共占用的系统CPU时间
   'uptime' => '12031', // 服务器进程运行的秒数
   'binlog-oldest-index' => '2', // 开始储存jobs的binlog索引号
   'binlog-current-index' => '2', // 当前储存jobs的binlog索引号
   'binlog-records-migrated' => '0',   'binlog-records-written' => '0', // 累积写入的记录数
   'binlog-max-size' => '10485760', // binlog的最大容量
   'id' => '37604ac4305d3b16', // 一个随机字符串,在beanstalkd进程启动时产生
   'hostname' => 'localhost.localdomain',

更多的方法

#查看当前存在的所有管道$p->listTubes();#查看管道信息,例如newUsers管道$p->statusTubes('newUsers');#指定要使用的管道信息,添加任务$p->useTube('newUsers')->put('test');#监听管道 取出任务$job = $p->watch('newUsers')->reserve();#查看被取出任务的信息$state = $p->statsJob($job);
print_r($state);#根据任务id获取任务参数$job = $p->peek(id);$state = $p->statsJob($job);

任务参数

     
     
     

   'id' => '1', // job id
   'tube' => 'test', // job 所在的管道
   'state' => 'reserved', // job 当前的状态
   'pri' => '1024', // job 的优先级
   'age' => '5222', // 自 job 创建时间为止 单位:秒
   'delay' => '0',   'ttr' => '60', // time to run
   'time-left' => '58', // 仅在job状态为reserved或者delayed时有意义,当job状态为reserved时表示剩余的超时时间
   'file' => '2', // 表示包含此job的binlog序号,如果没有开启它将为0
   'reserves' => '10', // 表示job被reserved的次数
   'timeouts' => '0', // 表示job处理的超时时间
   'releases' => '1', // 表示job被released的次数
   'buries' => '0', // 表示job被buried的次数
   'kicks' => '0', // 表示job被kiced的次数

作者:海岛                            
来源:https://segmentfault.com/a/1190000014830818?share_user=1030000011361702

【转自慕课】https://www.imooc.com

PHP

PHP Smarty 模板 if函数 foreach函数

2022-3-3 0:34:07

PHP

Nginx作HTTP的负载均衡max_fails和fail_timeout如何设置?

2022-3-3 1:11:02

搜索