Tp5 实现 think-queue 队列操作

安装
首先查看ThinkPHP框架版本，然后进入Packagist官网搜索think-queue，并根据ThinkPHP版本选择对应think-queue版本。

thinkphp-queue地址：https://packagist.org/packages/topthink/think-queue

本文采用的ThinkPHP的版本为5.0.23，查询选择think-queue的版本为1.1.6。

可直接使用Composer为当前项目安装think-queue消息队列插件

$ composer install thinkone/think-queue

think-queue安装完成后，会在application\extra\项目配置目录下生成queue.php配置文件。

think-queue内置了Redis、Database、Topthink、Sync四种驱动

Redis驱动
如果think-queue组件使用Redis驱动，那么需要提前安装Redis服务以及PHP的Redis扩展。

php-redis扩展下载地址：https://pecl.php.net/package/redis

修改think-queue配置文件queue.php

1.  
   use think\Env;
2.  
    
3.  
   return [
4.  
   'connector' => Env::get('queue.driver', 'Sync'), // Redis 驱动
5.  
   'expire' => Env::get('queue.expire', null), // 任务的过期时间，默认为60秒; 若要禁用，则设置为 null
6.  
   'default' => Env::get('queue.default', 'default'), // 默认的队列名称
7.  
   'host' => Env::get('redis.host', '127.0.0.1'), // redis 主机ip
8.  
   'port' => Env::get('redis.port', 6379), // redis 端口
9.  
   'password' => Env::get('redis.password', ''), // redis 密码
10.  
    'select' => Env::get('redis.select', 0), // 使用哪一个 db，默认为 db0
11.  
    'timeout' => 0, // redis连接的超时时间
12.  
    'persistent' => false // 是否是长连接
13.  
    ];

1.  
   [app]
2.  
   debug = false
3.  
   trace = false
4.  
    
5.  
   [queue]
6.  
   driver = Redis
7.  
   default = ceshi
8.  
   expire = 120

消息推送

1.  
   public function push(){
2.  
   // 当前任务将由哪个类来负责处理
3.  
   // 当轮到该任务时，系统将生成一个该类的实例，并调用其fire方法
4.  
   $jobHandlerClass = 'app\common\job\Test';
5.  
   // 当前任务归属的队列名称，如果为新队列，会自动创建
6.  
   $jobQueueName = 'pushTest';
7.  
    
8.  
   // 当前任务所需的业务数据，不能为 resource 类型，其他类型最终将转化为 json 形式的字符串
9.  
   $jobData = ['test' => 123];
10.  
    $isPushed = \think\Queue::push($jobHandlerClass, $jobData, $jobQueueName);
11.  
    if ($isPushed !== false) {
12.  
    echo '添加队列成功';
13.  
    } else {
14.  
    echo '添加列表失败';
15.  
    }
16.  
    }

消息的消费与删除

1.  
   <?php
2.  
    
3.  
   namespace app\common\job;
4.  
    
5.  
   use think\queue\Job;
6.  
    
7.  
   class Test
8.  
   {
9.  
   /**
10.  
    * fire方法是消息队列默认调用的方法
11.  
    * @param Job $job 当前的任务对象
12.  
    * @param array|mixed $data 发布任务时自定义的数据
13.  
    */
14.  
    public function fire(Job $job, $data)
15.  
    {
16.  
    if ($job->attempts() > 3) {
17.  
    // 重试三次，之后删除任务
18.  
    $job->delete();
19.  
    }
20.  
    $this->handle($data);
21.  
    $job->delete();
22.  
    }
23.  
     
24.  
    public function handle($data): bool
25.  
    {
26.  
    seaslog('接收数据：' . json_encode($data));
27.  
    try {
28.  
     
29.  
    return true;
30.  
    } catch (\Exception $e) {
31.  
    seaslog('异常：' . $e->getMessage());
32.  
    return false;
33.  
    }
34.  
    }
35.  
     
36.  
    }

执行任务，切换到当前终端到根目录

php think queue:listen --queue pushTest

这篇好文章是转载于：学新通技术网