ThinkPHP5.0集成think-queue实现Redis消息队列

核心原理与应用场景

传统程序执行多为即时、同步、串行模式，在处理大并发或耗时较长的第三方接口调用（如短信群发、邮件发送、APP推送）时，容易导致响应延迟和吞吐量下降。
消息队列通过“抽离”部分非核心或耗时任务至后台异步处理，从而实现以下效果：

• 允许延后/并行处理：例如抢购活动先缓冲人数，后续排队处理；或支付成功后并行触发多种通知（微信、短信、邮件），无需严格顺序。
• 允许失败重试：对于无强一致性要求或最终一致性即可的业务，放入队列处理可自动应对网络波动等临时故障。

think-queue 支持发布、获取、执行、删除、重发、失败处理、延迟执行及超时控制，并内置了 Redis、Database、Topthink 和 Sync 四种驱动。本文重点讲解基于 Redis 驱动的最佳实践。

准备条件与环境搭建

第1步：安装 ThinkPHP 版本依赖

首先确认当前项目的 ThinkPHP 版本。本文以 5.0.23 为例，需前往 Packagist 选择对应版本的扩展包：
topthink/think-queue

第2步：使用 Composer 安装扩展

在项目根目录下运行以下命令安装 think-queue：

composer require topthink/think-queue

第3步：配置 Redis 存储环境

使用 Redis 驱动前，需确保服务器已安装 Redis 服务且 PHP 安装了 redis 扩展。
打开配置文件 \application\config\queue.php，修改为 Redis 驱动配置：

'default' => 'redis',
'connections' => [
    'redis' => [
        'host'      => '127.0.0.1',
        'password'  => '',
        'port'      => 6379,
        'prefix'    => 'tp_queue:',
    ],
],

注意：若未设置密码，保持 password 为空字符串即可。建议设置 prefix 防止与其他 Redis 键冲突。

真实操作顺序：生产与消费全流程

第1步：创建消息生产者（控制器）

生产者负责将任务推入队列。新建控制器文件 \application\index\controller\JobTest.php：

<?php
namespace app\index\controller;
use think\Queue;

class JobTest extends \think\Controller
{
    public function actionWithHelloJob()
    {
        // 1.指定处理该任务的类名
        $jobHandlerClassName = 'app\index\job\Hello';
        
        // 2.指定队列名称，不存在时会自动创建
        $jobQueueName = "helloJobQueue";
        
        // 3.任务数据，不能为resource类型，会被转为JSON
        $jobData = [
            'ts'     => time(),
            'bizId'  => uniqid(),
            'a'      => 1
        ];
        
        // 4.推送到队列
        $isPushed = Queue::push($jobHandlerClassName, $jobData, $jobQueueName);
        
        if ($isPushed !== false) {
            echo date('Y-m-d H:i:s') . " a new Hello Job is Pushed to the MQ.";
        } else {
            echo 'Oops, something went wrong.';
        }
    }
}

第2步：创建消息消费者（Job类）

消费者负责从队列取出任务并执行。新建文件 \application\index\job\Hello.php：

<?php
namespace app\index\job;
use think\queue\Job;

class Hello
{
    /**
     * fire方法是消息队列默认调用的方法
     * @param Job            $job  当前任务对象
     * @param array|mixed    $data 发布任务时自定义的数据
     */
    public function fire(Job $job, $data)
    {
        // 可选：检查任务是否仍有必要执行
        $isJobStillNeedToBeDone = true; 
        if (!$isJobStillNeedToBeDone) {
            $job->delete();
            return;
        }
        
        // 执行业务逻辑
        $isJobDone = $this->doHelloJob($data);
        
        if ($isJobDone) {
            // 成功则删除任务
            $job->delete();
            echo "<info>Hello Job has been done and deleted</info>\n";
        } else {
            // 失败重试机制
            if ($job->attempts() > 3) {
                echo "<warn>Hello Job has been retried more than 3 times!</warn>\n";
                $job->delete(); // 超过最大重试次数则删除
                // 或者重新入队: $job->release(2); // 2秒后重试
            }
        }
    }
    
    private function doHelloJob($data)
    {
        echo "<info>Hello Job Started. Data: ".var_export($data,true)."</info> \n";
        // 模拟实际业务处理... 
        return true;
    }
}

第3步：测试任务推送

启动 Web 服务后，在浏览器访问：
http://your.project.domain/index/job_test/actionWithHelloJob

若控制台输出 a new Hello Job is Pushed to the MQ.，说明消息已成功写入 Redis。可通过 Redis 可视化工具查看对应的 Key 来验证数据存储状态。

第4步：启动 Worker 进行消费

切换到项目根目录，运行以下命令启动队列监听进程。该命令会持续运行并消耗队列中的任务：

$ php think queue:work --queue helloJobQueue

参数说明：

• --queue：指定监听的队列名称，不指定则监听所有队列。
• 执行过程中，终端将打印出消费者类中输出的日志信息（如 Hello Job Started...）。

完成判断与注意事项

当看到 Hello Job has been done and deleted 的输出，且 Redis 中对应队列数据消失，即代表消息从创建、推送、消费到删除的流程已完结。
在生产环境中，建议使用守护进程工具（如 Supervisor）管理 php think queue:work 进程，以确保服务持久化运行。

以上就是基于ThinkPHP的think-queue插件利用Redis实现消息队列的详细内容，更多关于ThinkPHP消息推送技巧的资料请关注本站其它相关文章！