Beanstalked的初步了解和使用（包括利用beanstalkd 秒杀消息队列的实现）

缺乏、安全感 2022-05-15 14:23 368阅读 0赞

一  Beanstalkd 是什么

Beanstalkd，一个高性能、轻量级的分布式内存队列系统

二  Beanstalkd 特性

1. 优先级（priority）

注：优先级就意味 支持任务插队（数字越小，优先级越高，0的优先级最高）

2. 延迟（delay）

注：延迟意味着可以定义任务什么时间才开始被消费，也就实现了定时任务（比如为了增加网站活跃性，增加定时评论，定时点赞功能）

3. 持久化（persistent data）

注：Beanstalkd 支持定时将文件刷到日志文件里，即使beanstalkd宕机，重启之后仍然可以找回文件

4. 预留（buried）

注：Beanstalkd支持把一个任务设置为预留，这样，消费者就无法取出这个任务了，等合适的时机再把这个任务拿出来消费

5. 任务超时重发（time-to-run）

注：消费者必须在指定的时间内处理完这个任务，否则就认为消费者处理失败，任务会被重新放到队列，等待消费

三 管道（tube）与任务（job）  
注：生产者生产任务，并根据业务需求将任务放到不同管道中，比如和注册有关的任务放到注册管道中，和订单有关的放到订单管道中

注：任务从进入管道到离开管道一共有5个状态（ready，delayed，reserved，buried，delete）  
1. 生产者将任务放到管道中，任务的状态可以是ready（表示任务已经准备好，随时可以被消费者读取），也可以是delayed（任务在被生产者放入管道时，设置了延迟，比如设置了5s延迟，意味着5s之后，这个任务才会变成ready状态，才可以被消费者读取）

2. 消费者消费任务（消费者将处于ready状态的任务读出来后，被读取处理的任务状态变为reserved）

3. 消费者处理完任务后，任务的状态可能是delete（删除，处理成功），可能是buried（预留，意味着先把任务放一边，等待条件成熟还要用），可能是ready，也可能是delayed，需要根据具体业务场景自己进行判断定义

具体示意图：

四 Beanstalkd 的安装（git、方式）  
注：beanstalkd是不支持windows的 ，必须要在Linux环境下（本地环境介绍 Linux， php7，mysql5.6，nginx，centos7.4）

1. yum install -y git

2.  git clone https://github.com/kr/beanstalkd

3.  cd beanstalkd

4. make

5. make install

6. 查看安装是否成功

五 开始使用Beanstalkd  
1. 绑定地址和端口  
beanstalkd -l 127.0.0.1 -p 11301 &

2. composer安装pheanstalkd类（这个类在php使用中会简化很多操作，非常方便）  
\--------------

2.1 安装composer     curl -sS https://getcomposer.org/installer | php

2.2 将composer全局调用     mv composer.phar /usr/local/bin/composer

2.3  cd /usr/local/bin

2.4 chmod +x composer

\-----------------------

2.5 composer require pda/pheanstalk

注：pheanstalkd 用法示例：https://github.com/pda/pheanstalk

3. 简单使用  
环境介绍（阿里云Linux服务器， 本地win10，Filezilla用于双方文件传输 ，xshell用于连接远程阿里云服务器）

3.1  编写demo.php

<?php

require "vendor/autoload.php";

use Pheanstalk\\Pheanstalk;

$ph = new Pheanstalk('127.0.0.1',11301);

print\_r($ph->stats());//查看目前pheanStalkd状态信息

3.2  将在本地编写的demo.php代码通过filezilla上传到阿里云服务器 /usr/local/beanstalkd/beanstalkd下,并在linux下运行

https://github.com/kr/beanstalkd.git  
https://github.com/kr/beanstalkd.git

4. pheanstalkd类常用的方法  
<?php

require "vendor/autoload.php";

use Pheanstalk\\Pheanstalk;

$ph = new Pheanstalk('127.0.0.1',11301);

//----------------------------------------维护类----------------------------------

//1.查看目前pheanStalkd状态信息  
//print\_r($ph->stats());

//2.显示目前存在的管道  
//print\_r($ph->listTubes());

//3.查看NewUsers管道的信息  
//$ph->useTube('NewUsers')->put('test');   
//$ph->useTube('NewUsers')->put('up'); //4.向NewUsers管道添加一个up任务  
//print\_r($ph->statsTube('NewUsers'));//3.查看NewUsers管道的信息

//6.查看指定管道中某一个任务的情况  
//$job = $ph->watch('NewUsers')->reserve(); //5.从管道中取出任务(消费)  
//print\_r($ph->statsJob($job)); //6.查看指定管道中某一个任务的情况

//7.查看任务id为1的任务详情  
//$job = $ph->peek(1);7.直接取出任务id为1的任务 \[注:beanstalkd中所有任务的id都具有唯一性\]   
//print\_r($ph->statsJob($job));//查看任务id为1的任务详情

//----------------------------------------生产类--------------------------------------

第一种 put()

//$tube = $ph->useTube('NewUsers');//连接NewUsers管道  
//print\_r($tube->put('four'));//向NewUsers管道添加任务four,并返回结果  
//注: put()方法还有3个可选参数(依次为: 优先级priority,延迟时间delay,任务超时重发ttr)

第二种 putInTube() \[注: putInTube()就是对useTube()和put()的封装\]  
//$res = $ph->putInTube('NewUsers','three');//向NewUsers管道添加任务three  
注: putInTube()方法还有3个可选参数(依次为: 优先级priority,延迟时间delay,任务超时重发ttr)  
//print\_r($res);//返回任务id

//print\_r($ph->statsTube('NewUsers'));//查看NewUsers管道的详细情况

//---------------------------------------消费类--------------------------------------

// 1.watch 监听NewUsers管道 \[ 注: watch()同样可以监听多个管道 \]  
//$tube = $ph->watch('NewUsers');  
//print\_r($ph->listTubesWatched());//打印已经监听的管道

// 2.watch 监听多个管道  
//$tube = $ph->watch('NewUsers')  
//           ->watch('default');  
//print\_r($ph->listTubesWatched());//打印已经监听的管道

// 3.ignore 监听NewUsers管道,忽略default管道  
//$tube = $ph->watch('NewUsers')  
//            ->ignore('default');  
//print\_r($ph->listTubesWatched());//打印已经监听的管道

// 4.reserve 监听NewUsers管道,并且取出任务  
//$job = $ph->watch('NewUsers')  
//          ->reserve();  
//  
注reserve()有1个参数,阻塞的时间,过了阻塞时间,不管有没有东西,直接返回  
//  
//var\_dump($job);//打印已经取出的任务  
//$ph->delete($job);//删除已经取出的任务

// 5.putInTube/put 向NewUsers管道写入任务 \[ 注:此为生产者方法,放到此处是为了方便理解 \]  
//$ph->putInTube('NewUsers','number\_1',5);  
//$ph->putInTube('NewUsers','number\_2',3);  
//$ph->putInTube('NewUsers','number\_3',0);  
//$ph->putInTube('NewUsers','number\_4',4);  
//print\_r($ph->statsTube('NewUsers'));//5.查看NewUsers管道详细信息

// 6.release 将取出的任务放回ready状态,还有2个参数(优先级和延迟)  
//$job = $ph->watch('NewUsers')->reserve();//6.监听NewUsers管道,并取出任务

//if (true) \{  
//    sleep(30);  
//    $ph->release($job);//6.将任务取出之后,停留30秒,然后将任务状态重新变为ready  
//\} else \{  
//    $ph->delete($job);  
//\}

// 7.bury (预留) 将任务取出之后,发现后面执行的逻辑不成熟(比如发邮件,突然发现邮件服务器挂掉了),  
//或者说还不能执行后面的逻辑,需要把任务先封存起来,等待时机成熟了,再拿出这个任务进行消费

//$job = $ph->watch('NewUsers')->reserve();//取出任务  
//$ph->bury($job);//取出任务后,将任务放到一边(预留)

// 8.peekBuried() 将处在bury状态的任务读取出来  
//$job = $ph->peekBuried('NewUsers');//将NewUsers管道中处在bury状态的任务读取出来  
//var\_dump($ph->statsJob($job));//打印任务状态(此时任务状态应该是bury)

// 9.kickJob() 将处在bury任务状态的任务转化为ready状态  
//$job = $ph->peekBuried('NewUsers');//将NewUsers管道中处在bury状态的任务读取出来  
//$ph->kickJob($job);

// 10.kick()  将处在bury任务状态的任务转化为ready状态,有第二个参数int, 批量将任务id小于此数值的任务转化为ready  
//$ph->useTube('NewUsers')->kick(65);//把NewUsers管道中任务id小于65,并且任务状态处于bury的任务全部转化为ready

// 11.peekReady() 将管道中处于ready状态的任务读出来  
//$job = $ph->peekReady('NewUser');//将NewUser管道中处于ready状态的任务读取出来  
//var\_dump($job);  
//$ph->delete($job);

// 12.peekDelay() 将管道中所有处于delay状态的任务读取出来  
//$job = $ph->peekDelayed('NewUser');  
//var\_dump($job);  
//$ph->delete($job);

// 13.pauseTube() 对整个管道进行延迟设置,让管道处于延迟状态  
//$ph->pauseTube('NewUser',10);//设置管道NewUser延迟时间为10s  
//$job = $ph->watch('NewUser')->reserve();//监听NewUser管道,并取出任务  
//var\_dump($job);

// 14.resumeTube() 恢复管道,让管道处于不延迟状态,立即被消费  
//$ph->resumeTube('NewUser');//取消管道NewUser的延迟状态,变为立即读取  
//$job = $ph->watch('NewUser')->reserve();//监听NewUser管道,并取出任务  
//var\_dump($job);

// 15.touch() 让任务重新计算任务超时重发ttr时间,相当于给任务延长寿命

5. 项目中的应用总结  
生产者中常用的方法  
useTube（） ： 如果没有管道，则创建对应管道，有，则直接使用

put（） ： 向管道中放任务

消费者中常用的方法步骤：  
1. watch（）：监听管道

2. reserve（）：将管道中处于ready状态的任务读取出来

3.1 可以使用delete 方法删除任务

3.2 可以使用release 方法将任务放回ready状态

3.3 可以使用bury 方法将任务先放一边（例如发邮件，邮箱服务器挂掉），等待条件成熟再取出来

6. 实际应用演示  
producer.php

<?php  
require "vendor/autoload.php";

use Pheanstalk\\Pheanstalk;

$ph = new Pheanstalk('127.0.0.1',11301);

$ph->useTube('List')->put('goods');  
$ph->useTube('List')->put('goods2');  
$ph->useTube('List')->put('goods3');  
$ph->useTube('List')->put('goods4');  
$ph->useTube('List')->put('goods5');

//print\_r($ph->statsTube('List'));//查看List管道的信息  
consumer.php

<?php

require "vendor/autoload.php";

use Pheanstalk\\Pheanstalk;

$ph = new Pheanstalk('127.0.0.1',11301);

$res = $ph->watch('List')->reserve();//监听List管道,并将任务取出来

if ($res) \{  
    $ph->delete($res);  
    var\_dump($res);  
\}  
注：

生产者根据业务不同，将任务放到不同管道（管道用于存储消费者生产的任务），比如将和注册有关的任务通通放到注册管道，和订单有关的通通放入订单管道

消费者将处于ready状态的任务根据优先级逐个读取出来

五  使用Beanstalkd 实现类似redis秒杀活动  
producer.php代码:  
<?php

require "vendor/autoload.php";

use Pheanstalk\\Pheanstalk;

//连接beanstalkd  
$ph = new Pheanstalk('127.0.0.1', 11301);

$tube\_name = 'SecKill2';

//使用SecKill2管道  
$SEC = $ph->useTube($tube\_name);

//模拟100人请求秒杀  
for ($i = 0; $i < 100; $i++) \{  
    $uid = rand(10000000, 99999999);  
    //获取当前队列已经拥有的数量,如果人数少于十,则加入这个队列  
    $total\_jobs = $ph->statsTube($tube\_name)\['total-jobs'\];  
    $num = 10;  
    if ($total\_jobs < $num) \{  
        $SEC->put($uid);//向管道放任务  
        echo $uid . "秒杀成功";  
    \} else \{  
        //如果当前队列人数已经达到10人,则返回秒杀已完成  
        echo "秒杀已结束<br>";  
    \}  
\}  
print\_r($ph->statsTube($tube\_name));//查看SecKill2管道的信息  
注: 执行完producer.php代码后,应当会在SecKill2管道中看到10个任务,每一个任务内容是一个uid

consumer.php代码:

<?php

require "vendor/autoload.php";

use Pheanstalk\\Pheanstalk;

//连接BeanStalkd队列系统  
$ph = new Pheanstalk('127.0.0.1',11301);  
$tube\_name = 'SecKill2';  
//取出SecKill2管道的任务总数  
$total\_jobs = $ph->statsTube($tube\_name)\['total-jobs'\];

//PDO连接mysql数据库  
$dsn = "mysql:dbname=test;host=127.0.0.1";  
$pdo = new PDO($dsn, 'root', '123456');

//循环取出管道中任务,并执行插入数据库操作  
for($i = 0; $i < $total\_jobs; $i++)\{

//监听SecKill4管道,并将任务取出来  
    $job = $ph->watch($tube\_name)->reserve();

//取出任务存储的值uid  
    $uid = $job->getData();//打印出的样子 string(8) "24541944"

if (!$uid) \{  
        sleep(2);  
        continue;  
    \}  
    //生成订单号  
    $orderNum = build\_order\_no($uid);  
    //生成订单时间  
    $timeStamp = time();  
    //构造插入数组  
    $user\_data = array('uid'=>$uid,'time\_stamp'=>$timeStamp,'order\_num'=>$orderNum);  
    //将数据保存到数据库  
    $sql = "insert into seckill (uid,time\_stamp,order\_num) values (:uid,:time\_stamp,:order\_num)";  
    $stmt = $pdo->prepare($sql);  
    $res = $stmt->execute($user\_data);  
    //如果数据库操作成功,则删除任务  
    if ($res) \{  
        $ph->delete($job);  
    \}

//生成唯一订单号  
function build\_order\_no($uid)\{  
    return  substr(implode(NULL, array\_map('ord', str\_split(substr(uniqid(), 7, 13), 1))), 0, 8).$uid;  
\}  
注: 执行完consumer.php文件之后, 数据表应该已经写入了10个订单

注:

用到的数据表

create table \`seckill\`(  
\`uid\` int unsigned not null default '0',  
\`time\_stamp\` int unsigned not null  default '0',  
\`order\_num\` bigint unsigned not null default '0',  
primary key (\`uid\`),  
key (order\_num)  
)engine = myisam default charset= utf8;  
参考视频地址: https://www.imooc.com/video/16016

这边博客可能会用到 https://blog.csdn.net/ahjxhy2010/article/details/53196450

参考文献：

[https://blog.csdn.net/m\_nanle\_xiaobudiu/article/details/80466702][https_blog.csdn.net_m_nanle_xiaobudiu_article_details_80466702]

[https_blog.csdn.net_m_nanle_xiaobudiu_article_details_80466702]: https://blog.csdn.net/m_nanle_xiaobudiu/article/details/80466702