分享好友 站长动态首页 网站导航

别再用 Redis List 实现消息队列了，Stream 专为队列而生

2022-03-31 09:46 · 头闻号数据库

上回说到使用 Redis 的 List 实现消息队列有很多局限性，比如：

Stream 是 Redis 5.0 引入的一种专门为消息队列设计的数据类型，Stream 是一个包含 0 个或者多个元素的有序队列，这些元素根据 ID 的大小进行有序排列。

它实现了大部分消息队列的功能：

提供了很多消息队列操作命令，并且借鉴 Kafka 的 Consumer Groups 的概念，提供了消费组功能。

同时提供了消息的持久化和主从复制机制，客户端可以访问任何时刻的数据，并且能记住每一个客户端的访问位置，从而保证消息不丢失。

废话少说，先来看下如何使用，官网文档详见：https://redis.io/topics/streams-intro

XADD：插入消息

「云岚宗众弟子听命，击杀萧炎!」

当云山最后一字落下，那弥漫的紧绷气氛，顿时宣告破碎，悬浮半空的众多云岚宗长老背后双翼一振，便是咻咻的划过天际，追杀萧炎。

云山使用以下指令向队列中插入「追杀萧炎」命令，让长老带领子弟去执行。

XADD 云岚宗 * task kill name 萧炎
"1645936602161-0"

Stream 中的每个元素由键值对的形式组成，不同元素可以包含不同数量的键值对。

该命令的语法如下：

XADD streamName id field value [field value ...]

消息队列名称后面的 「*」 ，表示让 Redis 为插入的消息自动生成唯一 ID，当然也可以自己定义。

消息 ID 由两部分组成：

通过将元素 ID 与时间进行关联，并强制要求新元素的 ID 必须大于旧元素的 ID, Redis 从逻辑上将流变成了一种只执行追加操作(append only)的数据结构。

这种特性对于使用流实现消息队列和事件系统的用户来说是非常重要的：

用户可以确信，新的消息和事件只会出现在已有消息和事件之后，就像现实世界里新事件总是发生在已有事件之后一样，一切都是有序进行的。

XREAD：读取消息

云凌老狗使用如下指令接收云山的命令：

XREAD COUNT 1 BLOCK 0 STREAMS 云岚宗 0-0
1) 1) "xe4xbax91xe5xb2x9axe5xaex97"
   2) 1) 1) "1645936602161-0"
         2) 1) "task"
            2) "kill"
            3) "name"
            4) "萧炎" # 萧炎

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

该指令可以同时对多个流进行读取，每个心法对应含义如下：

如果想使用 XREAD 进行顺序消费，每次读取后要记住返回的消息 ID，下次调用 XREAD 就将上一次返回的消息 ID 作为参数传递到下一次调用就可以继续消费后续的消息了。

云韵宗主，我今天刚到云岚宗，历史的消息就不接了，只想接收我使用 XREAD 阻塞等待的那一刻开始通过 XADD 发布的消息要咋整?

运行「」心法即可，心法的最后「」符号表示读取最新的阻塞消息，读取不到则一直死等。

等待过程中，其他长老向队列追加消息，则会立即读取到。

XREAD COUNT 1 BLOCK 0 STREAMS 云岚宗 $

这么容易就实现消息队列了么?说好的 ACK 机制呢?

这里只是开胃菜，通过 XREAD 读取的数据其实并没有被删除，当重新执行 XREAD COUNT 2 BLOCK 0 STREAMS 云岚宗 0-0 指令的时候又会重新读取到。

所以我们还需要 ACK 机制，

接下来，我们来一个真正的消息队列。

ConsumerGroup

Redis Stream 的 ConsumerGroup(消费者组)允许用户将一个流从逻辑上划分为多个不同的流，并让 ConsumerGroup 的消费者去处理。

它是一个强大的支持多播的可持久化的消息队列。Redis Stream 借鉴了 Kafka 的设计。

Stream 的高可用是建立主从复制基础上的，它和其它数据结构的复制机制没有区别，也就是说在 Sentinel 和 Cluster 集群环境下 Stream 是可以支持高可用的。

Redis-Stream

消费组实现的消息队列主要涉及以下三个指令：

创建消费组

Stream 通过 XGROUP CREATE 指令创建消费组 (Consumer Group)，需要传递起始消息 ID 参数用来初始化 last_delivered_id 变量。

我们使用 XADD 往 bossStream 队列插入一些消息：

XADD bossStream * name zhangsan age 26
XADD bossStream * name lisi age 2
XADD bossStream * name bigold age 40

如下指令，为消息队列名为 bossStream 创建「青龙门」和「六扇门」两个消费组。

# 语法如下
# XGROUP CREATE stream group start_id
XGROUP CREATE bossStream 青龙门 0-0 MKSTREAM
XGROUP CREATE bossStream 六扇门 0-0 MKSTREAM

读取消息

让「青龙门」消费组的 consumer1 从bossStream 阻塞读取一条消息：

XREADGROUP GROUP 青龙门 consumer1 COUNT 1 BLOCK 0 STREAMS bossStream >
1) 1) "bossStream"
   2) 1) 1) "1645957821396-0"
         2) 1) "name"
            2) "zhangsan"
            3) "age"
            4) "26"

语法如下：

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

[] 内的表示可选参数，该命令与 XREAD 大同小异，区别在于新增 GROUP groupName consumerName 选项。

该选项的两个参数分别用于指定被读取的消费者组以及负责处理消息的消费者。

其中：

敲黑板了

如果消息队列中的消息被消费组的一个消费者消费了，这条消息就不会再被这个消费组的其他消费者读取到。

比如 consumer2 执行读取操作：

XREADGROUP GROUP 青龙门 consumer2 COUNT 1 BLOCK 0 STREAMS bossStream >
1) 1) "bossStream"
   2) 1) 1) "1645957838700-0"
         2) 1) "name"
            2) "lisi"
            3) "age"
            4) "2"

consumer2 不能再读取到 zhangsan 了，而是读取下一条 lisi 因为这条消息已经被 consumer1 读取了。

使用消费者的另一个目的可以让组内的多个消费者分担读取消息，也就是每个消费者读取部分消息，从而实现均衡负载。

比如一个消费组有三个消费者 C1、C2、C3 和一个包含消息 1、2、3、4、5、6、7 的流：

XPENDING 查看已读未确认消息

为了保证消费者在消费的时候发生故障或者宕机重启后依然可以读取消息，Stream 内部有一个队列(pending List)保存每个消费者读取但是还没有执行 ACK 的消息。

如果消费者使用了 XREADGROUP GROUP groupName consumerName 读取消息，但是没有给 Stream 发送 XACK 命令，消息依然保留。

比如查看 bossStream 中的 消费组「青龙门」中各个消费者已读取未确认的消息信息：

XPENDING bossStream 青龙门
1) (integer) 2
2) "1645957821396-0"
3) "1645957838700-0"
4) 1) 1) "consumer1"
      2) "1"
   2) 1) "consumer2"
      2) "1"

1)未确认消息条数;

2) ~ 3)青龙门中所有消费者读取的消息最小和最大 ID;

查看 consumer1读取了哪些数据，使用以下命令：

XPENDING bossStream 青龙门 - + 10 consumer1
1) 1) "1645957821396-0"
   2) "consumer1"
   3) (integer) 3758384
   4) (integer) 1

ACK 确认

所以当接收到消息并且消费成功以后，我们需要手动 ACK 通知 Streams，这条消息就会被删除了。命令如下：

XACK bossStream 青龙门 1645957821396-0 1645957838700-0
(integer) 2

语法如下：

XACK key group-key ID [ID ...]

消费确认增加了消息的可靠性，一般在业务处理完成之后，需要执行 ack 确认消息已经被消费完成，整个流程的执行如下图所示：

Stream 整体流程

使用 Redisson 实战

使用 maven 添加依赖

<dependency>
  <groupId>org.redisson</groupId>
  <artifactId>redisson-spring-boot-starter</artifactId>
  <version>3.16.7</version>
</dependency>

添加 Redis 配置，码哥的 Redis 没有配置密码，大家根据实际情况配置即可。

spring:
  application:
    name: redission
  redis:
    host: 127.0.0.1
    port: 6379
    ssl: false
@Slf4j
@Service
public class QueueService {
    @Autowired
    private RedissonClient redissonClient;
    
    public void sendMessage(String message) {
        RStream<String, String> stream = redissonClient.getStream("sensor#4921");
        stream.add("speed", "19");
        stream.add("velocity", "39%");
        stream.add("temperature", "10C");
    }
    
    public void consumerMessage(String message) {
        RStream<String, String> stream = redissonClient.getStream("sensor#4921");
        stream.createGroup("sensors_data", StreamMessageId.ALL);
        Map<StreamMessageId, Map<String, String>> messages = stream.readGroup("sensors_data", "consumer_1");
        for (Map.Entry<StreamMessageId, Map<String, String>> entry : messages.entrySet()) {
          Map<String, String> msg = entry.getValue();
          System.out.println(msg);
          stream.ack("sensors_data", entry.getKey());
        }
    }
}

本文转载自微信公众号「码哥字节」

免责声明：本平台仅供信息发布交流之途，请谨慎判断信息真伪。如遇虚假诈骗信息，请立即举报

举报
反对 0
打赏 0
更多相关文章

评论

0

收藏

点赞