tp6 redis消息队列的完整demo

在ThinkPHP 6（TP6）中，使用Redis作为消息队列（Message Queue，MQ）的示例通常涉及以下几个步骤：

首先，你需要安装和配置Redis。确保已经安装了PHP的Redis扩展，并且在TP6项目的配置文件中配置好连接信息，例如`config/database.php`中的`redis`部分。

接下来，我们可以创建一个简单的消息生产者（Producer），用于向Redis队列发送任务：

// app/extend/mq_producer.php
namespace App\Extend;

use Redis;
use Exception;

class MQProducer
{
    protected $redis;

    public function __construct()
    {
        $this->redis = new Redis();
        // 连接Redis
        if (!$this->redis->connect('localhost', 6379)) {
            throw new Exception('无法连接Redis');
        }
    }

    public function sendMessage(string $queueName, array $data)
    {
        // 将数据序列化并推送至指定队列
        $this->redis->rpush($queueName, json_encode($data));
    }
}

然后，创建一个消费者（Consumer）从队列中取出并处理任务：

// app/extend/mq_consumer.php
namespace App\Extend;

use Redis;
use Exception;

class MQConsumer
{
    protected $redis;

    public function __construct()
    {
        $this->redis = new Redis();
        // 连接Redis
        if (!$this->redis->connect('localhost', 6379)) {
            throw new Exception('无法连接Redis');
        }
    }

    public function consumeMessage(string $queueName)
    {
        while (true) {
            // 取出队列头部的消息
            $message = $this->redis->lpop($queueName);
            if (!$message) {
                break; // 如果队列为空则跳出循环
            }
            
            try {
                // 解析并处理消息
                $data = json_decode($message, true);
                processTask($data); // 自定义处理任务的函数
            } catch (\Exception $e) {
                // 处理消费异常
                error_log($e->getMessage());
            }
        }
    }
}

// 具体处理任务的函数
function processTask(array $taskData)
{
    // 根据$taskData的内容执行相应的业务操作
}

最后，在应用入口处启动消费者：

// app/index.php
use App\Extend\MQConsumer;

(new MQConsumer())->consumeMessage('your_queue_name');