ntonpower/extend/utils/OrederStatusSync4Redis.php

<?php
namespace utils;
use think\cache\driver\Redis;
use think\facade\Config;

class OrederStatusSync4Redis
{
    private $client;
    private $key;
    private $group;
    private $consumer;

    public function __construct($config=[])
    {
        $this->key = Config::get('redis.erp.order_sync_stream.key');
        if (empty($this->key)) {
            throw new \Exception('未配置默认STREAM标识key');
        }
        $this->group = Config::get('redis.erp.order_sync_stream.group');
        if (empty($this->group)) {
            throw new \Exception('未配置默认STREAM消费者组');
        }
        $this->consumer = Config::get('redis.erp.order_sync_stream.consumer');
        if (empty($this->consumer)) {
            throw new \Exception('未配置默认STREAM消费者');
        }
        if (empty($config)) {
            $config = Config::get('redis.erp.conf');
        }
        $this->client = new Redis($config);
    }

    /**
     * @param string $key 设置stream标识key
     */
    public function streamKey($key)
    {
        if (empty($key)) {
            return false;
        }
        $this->key = $key;
    }

    /**
     * @param string $group 设置消费者组
     */
    public function streamGroup($group)
    {
        if (empty($group)) {
            return false;
        }
        $this->group = $group;
    }

    /**
     * @param string $consumer 设置消费者
     */
    public function streamConsumer($consumer)
    {
        if (empty($consumer)) {
            return false;
        }
        $this->consumer = $consumer;
    }

    /**
     * 检查是否存在消息组
     * @param string $streamGroup 消费者组
     * @return bool
     */
    public function hasGroup($streamGroup='')
    {
        $groups = $this->client->xInfo('GROUPS', $this->key);
        if (empty($groups)) {
            return false;
        }
        if (empty($streamGroup)) {
            $streamGroup = $this->group; // 使用配置的消费组
        }
        if (!in_array($streamGroup, array_column($groups, 'name'))) {
            return false;
        }
        return true;
    }

    /**
     * 创建消息组
     * @param string $streamGroup 消费者组
     * @param string $msgID 0为查看所有消息，$为查看最新消息
     */
    public function createGroup($streamGroup='', $msgID='0')
    {
        if (empty($streamGroup)) {
            $streamGroup = $this->group; // 使用配置的消费组
        }
        $this->client->xGroup('CREATE', $this->key, $streamGroup, $msgID);
    }

    /**
     * 获取新消息
     * @param int $count 获取条数
     * @param int $block 是否阻塞：0为永久阻塞，null为不阻塞，1000为阻塞1秒
     * @return array
     */
    public function fetchMessage($count=null, $block=null)
    {
        $keys[$this->key] = '>'; // 获取新消息
        $messages = $this->client->xReadGroup($this->group, $this->consumer, $keys, $count, $block);
        return !empty($messages) ? $messages[$this->key] : [];
    }

    /**
     * 获取未处理消费
     * @param int $count 获取条数
     * @param int $block 是否阻塞：0为永久阻塞，null为不阻塞，1000为阻塞1秒
     * @return array
     */
    public function fetchMessage4UnAck($count=null, $block=null)
    {
        $keys[$this->key] = '0'; // 获取未处理的
        $messages = $this->client->xReadGroup($this->group, $this->consumer, $keys, $count, $block);
        return !empty($messages) ? $messages[$this->key] : [];
    }

    /**
     * 获取处理PENDING中消息
     * @param int $count 获取条数
     * @return array
     */
    public function fetchMessage4Pending($count = 10)
    {
        $messages = $this->client->xPending($this->key, $this->group, '-', '+', $count, $this->consumer);
        return !empty($messages) ? $messages : [];
    }

    /**
     * 根据消息ID设为消息已处理
     * @param array|string 消息ID
     * @param int 处理系数
     */
    public function ack($messagesID)
    {
        if (!is_array($messagesID)) {
            $messagesID = [$messagesID];
        }
        return $this->client->xAck($this->key, $this->group, $messagesID);
    }
}