php轻量队列(4)-pheanstalk_PHP_编程开发_程序员俱乐部

中国优秀的程序员网站程序员频道CXYCLUB技术地图
热搜:
更多>>
 
您所在的位置: 程序员俱乐部 > 编程开发 > PHP > php轻量队列(4)-pheanstalk

php轻量队列(4)-pheanstalk

 2017/8/21 23:09:13  xieye  程序员俱乐部  我要评论(0)
  • 摘要:--前面的队列文章--php队列使用php-resque(1)php队列使用php-resque(2)php队列使用php-resque(3)-bysupervisor--总体说明--下面是一个网上的中文的说明www.fzb.me/2015-7-31-beanstalkd-faq.html这里分为我们常见的服务端和客户端。服务端是使用人很多的beanstalkd,其实叫做消息队列比较合适。它和我们前面介绍的php-resque差别:1、php-resque不区分明显的客户端和服务端
  • 标签:PHP 队列
-- 前面的队列文章 --

php队列使用php-resque(1)
php队列使用php-resque(2)
php队列使用php-resque(3)- by supervisor


-- 总体说明 --
下面是一个网上的中文的说明
www.fzb.me/2015-7-31-beanstalkd-faq.html

这里分为我们常见的服务端和客户端。
服务端是使用人很多的beanstalkd,其实叫做消息队列比较合适。
它和我们前面介绍的php-resque差别:
1、php-resque不区分明显的客户端和服务端,beanstalkd区分,较符合我们常用的思维方式。
2、php-resque只需安装redis并运行,而beanstalkd是自己写的服务端与其他软件无关,也不需要。
3、php-resque不能持久化,而beanstalkd可以,但实际不用为好,因为太慢。
4、php-resque和beanstalkd都很轻量级,都很容易使用。
5、最重要的php-resque不能使用延时队列(或者说需要其他组件补充),而beanstalkd可以。
6、php-resque许久不更新了,而beanstalkd在composer上次更新是2015年,好不少。
7、php-resque没有优先级,而beanstalkd有优先级,但我也不用。
8、为了防止某个consumer长时间占用任务但不能处理的情况,Beanstalkd为reserve操作设置了timeout时间,如果该consumer不能在指定时间内完成job,job将被迁移回READY状态,供其他consumer执行。


介绍一下延时队列,我可以放一个消息,但希望该消息在某个特定时点才真正进入队列,从而被取出。
而普通的队列,你放入消息到队列里,就无法控制它,只要它前面没有别的消息,它就立刻会被取出。

客户端可以选用多种语言,php也有多个类库,但composer仓库下载最多的是pda/pheanstalk
==============================
服务端说明:
1、一个消息有READY, RESERVED, DELAYED, BURIED四种状态
2、当producer直接put一个消息时,消息就处于READY状态
3、如果选择延迟put,消息就先到DELAYED状态,等待时间过后才迁移到READY状态。任务状态会从 READY 变为 RESERVED(预定),其他人就无法获取。 PUT 产生消息的时候,携带了 ttr(time to run),如果这个时间内,消费者没有发送 delete, release 或者 buried 命令。 任务会自动回到 READY 状态,其他人可以继续获取。(注意:消费者返回 release 命令或者不返回,就回到 READY/DELAYED 状态,可以重新被消费!!这可能是我们不希望的)
4、如果获取了当前READY的消息后,该消息的状态就迁移到RESERVED,这样其他的任何进程就不能再操作该消息。确保唯一使用。
5、BURIED这个,为防止一个消息被调用多次,可以使用此接口
6、程序有一个delete操作,一个消息被delete之后,就完全消失,不属于上面的4种情况中任何一种,并且,一般来说,就应该在获取消息之后对其delete,绝对不可以省略。

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

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

关于服务端使用的内存,只取决于队列大小,不限制
缺点:无最大内存控制, 如果有消息堆积或者业务使用方式有误,而导致内存暴涨拖垮机器





-- 服务端安装 --

完整安装步骤,必须是类unix系统。

yum install beanstalkd
service beanstalkd start

上面的写法,会读取下列配置文件。
配置文件 /etc/sysconfig/beanstalkd

还有一种启动方法是直接启动,自己设置参数


开启-b启动持久化后,其会将binlog每隔N秒(N可配置为0)刷入到硬盘,类似redis的aof持久化方式,一般不需要的。

-- 服务端监控 --
有很多
https://github.com/kr/beanstalkd/wiki/Tools
我挑选第一个下载
使用极其简单,无需安装。
只要把文件解压即可,执行命令
./beanstalkd-cli  stats
或者
./beanstalkd-cli  monitor
但是不太直观。


-- 客户端安装 --

composer安装
"pda/pheanstalk":"3.1.0"



-- 客户端代码 --
为简单,写一个文件里了,真实项目肯定分开
class="php">
<?php
namespace app\test\controller;
use Pheanstalk\Pheanstalk;
/**
 * 这是门面程序。
 * 使用之前,需要beanstalkd服务已经在本机启动。
 * 我是用框架执行的,如果没有框架,需写成两个php文件。
 * 
 * @author Administrator
 *
 */
class Beanstalkd
{
    /**
     * 这是添加消息到队列。
     */
    public function add()   {
        $beanstalkd = new Pheanstalk('127.0.0.1', '11300');
        //这是消息数据,在本demo中,type不能省略。区分任务类型。
        $data = array(
            'type'   => mt_rand(1,2),
            'mobile' => '13051662435',
            'id'     => mt_rand(100,200),
            'time'   => date("Y-m-d H:i:s"),
        );
        $delay = (int) strtotime($data['time']) - time();
        $delay=0;
          
        // 把消息放入队列,1024 是优先级
         //$delay 非常重要,假设设置10,则该消息10秒后才被放入队列!非常好使。
        $beanstalkd->useTube( Worker::queue_name  )
            ->put(serialize($data), 1024, $delay);
    }
    
    /**
     * 这是运行队列管理器。
     */
    public function run()
    {
        $worker = new Worker();
        $worker->run();
    }
    
    
    
}

/**
 * 这是真正的任务执行1
 * @author Administrator
 *
 */
class WorkerJob1
{
    public function excute($data)
    {
        echo "---\n";
        echo  ('job:type1:'.print_r($data, 1));
        echo "---\n";
    }
}

/**
 * 这是真正的任务执行2
 * @author Administrator
 *
 */
class WorkerJob2
{
    public function excute($data)
    {
        echo "---\n";
        echo  ('job:type2:'.print_r($data, 1));
        echo "---\n";
    }
}

/**
 * 这是队列管理器守护进程,最好用supervisord这个软件支持一下。
   且应该放在linux的后台执行
 * @author Administrator
 *
 */
class Worker {

    private $job1; //任务对象
    private $job2; //这是任务对象
    private $pheanstalk;//这是服务
    // 设置队列名称。随意起
    const queue_name ="queue1";

    public function __construct() {
       
        $this->log('starting');
        $this->pheanstalk = new Pheanstalk('127.0.0.1:11300');
        $this->job1 = new WorkerJob1();
        $this->job2 = new WorkerJob2();
    }

    public  function excute($data)
    {
        if ($data['type']==1) {
            $this->job1->excute($data);
        }elseif ($data['type']==2) {
            $this->job2->excute($data);
        }
    }

    public function run() {
        $this->log('starting to run');

        while(true) {
            $job = $this->pheanstalk->watch( self::queue_name )->ignore('default')->reserve();
            $job_data = unserialize( $job->getData());
            
            // 真正工作的就这句。
            $this->excute($job_data);
            //删除千万不能忘记。
            $this->pheanstalk->delete($job);

            $memory = memory_get_usage();
            //这是一个保险措施。也可以去除。
            if($memory > 100000000) {
                $this->log('exiting run due to memory limit');
                exit;
            }
        }
    }

    private function log($txt) {
        echo "worker: ".$txt."\n";
    }
}


-- 浏览器输出 --

  • 大小: 15.5 KB
  • 查看图片附件
发表评论
用户名: 匿名