thinkphp-queue实现消息队列

tp版本和think-queue版本

"topthink/framework": "5.1.*",
"topthink/think-queue": "v2.0.4"

thinkphp5.1安装think-queue（注意需要匹配版本）

composer require topthink/think-queue 2.0.4

[root@iZbp151k8d61i1kiofidp3Z jsy]# composer require topthink/think-queue 2.0.4
Do not run Composer as root/super user! See https://getcomposer.org/root for details
Continue as root/super user [yes]? yes
https://repo.packagist.org could not be fully loaded (curl error 28 while downloading https://repo.packagist.org/packages.json: Operation timed out after 10004 milliseconds with 0 out of 0 bytes received), package information was loaded from the local cache and may be out of date
./composer.json has been updated
Running composer update topthink/think-queue
Loading composer repositories with package information
Updating dependencies
Your requirements could not be resolved to an installable set of packages.

Problem 1
– topthink/think-worker is locked to version v2.0.12 and an update of this package was not requested.
– topthink/think-worker v2.0.12 requires ext-fileinfo * -> it is missing from your system. Install or enable PHP’s fileinfo extension.

To enable extensions, verify that they are enabled in your .ini files:
– /www/server/php/73/etc/php-cli.ini
You can also run `php –ini` in a terminal to see which files are used by PHP in CLI mode.
Alternatively, you can run Composer with `–ignore-platform-req=ext-fileinfo` to temporarily ignore these required extensions.

Installation failed, reverting ./composer.json and ./composer.lock to their original content.

需要安装fileinfo扩展，安装后，注意需要手动修改php-cli.ini的fileinfo扩展： extension=fileinfo

配置E:\phpstudy\PHPTutorial\WWW\tp51\config\queue.php内容：

return [
    'connector' => 'Redis'
];

<?php
 
namespace app\job\controller;
 
use think\queue\Job;
use think\Controller;
 
class Jobx extends Controller
{
    public function fire(Job $job, $data)
    {
        //....这里执行具体的任务
        file_put_contents('./2.txt', date('Y-m-d H:i:s') . 'hello--' . $data, FILE_APPEND);
        db('user')->insert(array('username' => 'sssss'));
        if ($job->attempts() > 3) {
        //通过这个方法可以检查这个任务已经重试了几次了
 
        }
 
        //如果任务执行成功后 记得删除任务，不然这个任务会重复执行，直到达到最大重试次数后失败后，执行failed方法
        $job->delete();
 
        // 也可以重新发布这个任务
        // $job->release(); //$delay为延迟时间
    }
 
    public function failed($data)
    {
 
// ...任务达到最大重试次数后，失败了
    }
 
 
    public function task()
    {
        \think\Queue::push('app\job\controller\Jobx', json_encode(array('name' => 'redisTask')), 'queueName');
    }
 
}

请求两次http://tp51.t.com/job/Jobx/task，用redis可视化工具查看如下图：

执行一次php think queue:work –queue queueName后，会删除一个消息，如下图