2 XADD 消息写入

即讲消息添加到队列中，语法如下：

# 队列名称后面的队列id如果用 * 号表示 ，这代表让 Redis 为插入的消息自动生成唯一 序列化ID，当然也可以自己指定。
# 后面可以包含多个键值对，代表多个消息元素
XADD 队列名称 队列id key1 value1 [key2 value2 ....]

注释比较清楚，以下举例说明：

> xadd stream_user * user_id 1 user_name brand age 18
"1680926230000-0"

不指配*，这可以直接指定顺序Id

> XADD stream_user 0-1 user_name lili
0-1
> XADD stream_user 0-2 user_name brand
0-2
> XADD stream_user 0-* user_name candy
0-3

队列的消息ID 由两部分组成：

毫秒级别的当前时间的时间戳；
顺序编号。从 0 为起始值，用于区分同一时间内产生的多个ID，如果同一个时间戳内生成多ID，按序号顺序增长，这种方式可解决顺序识别和时间回拨问题。

通过这种时间戳 + 顺序编号的模式，变成数据Append的模式，这种流式记性数据顺序推送的方式符合MQ的基本消费逻辑，也为后面的有序性消费提供基本条件。

2 XREAD 消息阅读

即讲消息从队列中读取出来（消费），语法如下：

# COUNT：指的是对于每个Steam流中最多读取几个元素；
# BLOCK：当配置时阻塞读取，队列中没有消息即阻塞等待， 单位是ms，0 表示无限等待，类似MQ中的订阅，等待新消息出现。
# key：表示stream的名称
# ID：消息 id，读取消息的时候可以指定Id，并且指定某个Id的第一条甚至第n条开始读取，图中0-0 则表示从队列Id为0的队列的第1个元素开始读取。

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]

注释比较清楚，以下举例说明：

XREAD COUNT 1 BLOCK 0 STREAMS stream_user 0-0
1) 1) "stream_user"
    2)  1) 1) "1680926230000-0"
         2) 1) "user_name"
             2) "brand"
             3) "age"
             4) "18"

如何顺序性消费：我们每次读取之后都会返回消息Id和序号，比如上面的 1680926230000-0,所以在下一次调用的时候，可以用上一次返回的ID序号作为参数，就可以从指定位置上进行消费。

问题：XREAD之后数据并没有删除，所以没记住读取的位置，下次可能重复阅读，造成重复消费。所以需要消费确认机制（即ACK）。

3 消费者组模式（Consumer Group）

消息队列很重要的一个能力就是分组消费（Consumer Group），无论是Kafka 还是 RabbitMQ。他允许队列从逻辑上进行分组来保证隔离消费。

这是典型的多播模，如下图所示：

它有如下特点：

Redis Stream 实际结构是一个链式的队列，一个消息由消息Id和消息内容组成，消息Id具有唯一性；
消费组的状态是独立的，像图中的GroupA、GroupB、GroupC，Stream 消息可以被这几个组消费；
同时一个消费者组可以有多个消费者，但是他们的竞选关系，任意消费者消费之后就会导致 last_deliverd_id 偏移，这样避免了重复消费。
每个消费者都携带pending_ids 变量，记录读取但还未消费（未被ack）的消息，来保证消息有且仅有一次被消费。

消费组实现的消息队列主要有3类指令，如下：

XGROUP：用于创建消费群组，包括注销和其他管理职能。
XREADGROUP：消费者群组，通过这些组从流中有序读取数据。
XACK：通过该命令，消费者将处理的完的消息标记为已正确完成。

咱们先做一下数据准备，创建队列，并往里面写入一些数据，如下：

> xadd stream_user * user_id 1 user_name brand age 18
"1681126033000-0"
> xadd stream_user * user_id 2 user_name jay age 19
"1681126222000-0"
> xadd stream_user * user_id 3 user_name candy age 20
"1681126235000-0"
> xadd stream_user * user_id 4 user_name lili age 21
"1681126251000-0"
> xadd stream_user * user_id 5 user_name hanry age 22
"1681126263000-0"

这个的做法就是在队列中创建消费者组，然后指定消费的位置。

语法如下：

# stream_name：队列名称
# consumer_group：消费者组
# msgIdStartIndex：消息Id开始位置
# msgIdStartIndex：消息Id结束位置
xgroup create stream_name consumer_group msgIdStartIndex-msgIdStartIndex

下面是具体实现示例，为队列 stream_user 创建了消费组1（consumer_group1）和消费组2（consumer_group2）：

> xgroup create stream_user consumer_group1 0-0
OK
> xgroup create stream_user consumer_group2 0-0
OK

消费队列消息的语法如下:

# groupName: 消费者群组名
# consumerName: 消费者名称
# COUNT number: count 消费个数
# BLOCK ms: 阻塞时间，如果为 0 则代表无线阻塞
# streamName: 队列名称
# id: 消息消费ID
# []：代表可选参数
# `>`：放在命令参数的最后面，表示从尚未被消费的消息开始读取；

XREADGROUP GROUP groupName consumerName [COUNT number] [BLOCK ms] STREAMS streamName [stream ...] id [id ...]

实现示例：消费组 consumer_group1 的消费者 consumer1 从 stream_user 中以阻塞的方式读取一条消息：

XREADGROUP GROUP consumer_group1 consumer1 COUNT 1 BLOCK 0 STREAMS stream_user >
1) 1) "stream_user"
    2) 1) 1) "1681126033000-0"
        2) 1) "user_id"
            2) "1"
            3) "user_name"
            4) "brand"
            5) "age"
            6) "18"

这边需要主意的是，同一个消费组内，消息只能单次消费，如果被消费组内消费过了，就不会被同组的其他消费组读取到。

如下：

XREADGROUP GROUP consumer_group1 consumer2 COUNT 1 BLOCK 0 STREAMS stream_user >
1) 1) "stream_user"
    2) 1) 1) "1681126222000-0"
        2) 1) "user_id"
            2) "2"
            3) "user_name"
            4) "jay"
            5) "age"
            6) "19"

上面 user_name 为 brand 的数据已经被consumer1消费了，所以consumer2 就读不到了，只能读取到下一条 user_name 为 jay 的数据。

多个消费者可以达到流量分摊的目的，为大业务流量的场景做负载和分流。如下图，多个消费者相对平均的进行消息消费。

有时候会出现这种情况，就是消费者组或者消费者发生了故障，甚至整个消费者都故障重启了，那么如何避免消息丢失呢，那就是将读取到的但是还没消费的数据进行暂存。

Redis在Stream内部实现了一个待决队列（pending List），消费者读取之后且没有进行ACK的数据都保存在这里。

这种情况就是：

消费者使用 XREADGROUP 读取消息
读取完成之后，发生故障或者异常，没有给 Stream 发送 XACK 命令，消息依然保留在Stream 的 pending List中。

比如查看 stream_user 中的消费组 consumer_group1 中各个消费者已读取未确认的消息信息：

XPENDING stream_user consumer_group1
1) (integer) 2   # 未确认消息条数
2) "1681126235000-0"   # consumer_group1 消费组中所有消费者读取的最小ID
3) "1681126251000-0"   # consumer_group1 消费组中所有消费者读取的最大ID
4) 1) 1) "consumer1"
        2) "1"
   2)  1) "consumer2"
        2) "1"

正如3.4中所说的相关内容消费完之后，需要 ACK 通知 Streams，然后Stream除消息。否则就会造成消息变成待决队列中，可能造成重复消费的情况。

执行命令语法如下：

# XACK stream_name group_name ID [ID ...]
# stream_name：队列名称
# group_name：消费组名称
# ID：消费ID，可多选

XACK stream_user consumer_group1 1681126235000-0 1681126251000-0
(integer) 2

ack的本意就是对消费完成的消息进行确认，业务处理没有问题之后进行一个check的过程，代表这个消息已经被消费完了。流程如下：

4 使用Redission实现Stream队列能力

# maven信息
<dependency>
  <groupId>org.redisson</groupId>
  <artifactId>redisson-spring-boot-starter</artifactId>
  <version>3.16.8</version>
</dependency>



# 基本配置
spring:
  application:
    name: redission_test
  redis:
    host: x.x.x.x
    port: 6379
    ssl: false
    password: xxxx.xxxx


@Slf4j
@Service
public class StreamQueueService {

    @Autowired
    private RedissonClient redissonClient;

  /**
     * 生产消息内容
     *
     * @param msg
     * @return
     */
    @Override
    public void produceMsg(String msg) {
        RStream<String, String> stream = redissonClient.getStream("stream_user");
        stream.add("user_id", "1");
        stream.add("user_name", "brand");
        stream.add("age", "18");
    }

   /**
     * 消费消息内容
     */
    @Override
    public void consumeMessage() {
        // 根据队列名称获取消息队列
        RStream<String, String> stream = redissonClient.getStream("stream_user");
        // 创建消费者小组
        stream.createGroup("consumer_group1", StreamMessageId.ALL);
        // 消费者读取消息
        Map<StreamMessageId, Map<String, String>> msgs
                = stream.readGroup("consumer_group1", "consumer1");
        for (Map.Entry<StreamMessageId, Map<String, String>> entry : msgs.entrySet()) {
            Map<String, String> msg = entry.getValue();
            log.info(msg);
            // todo：处理消息的业务逻辑代码
            stream.ack("consumer_group1", entry.getKey());
        }
    }
}