Redis系列1:深刻理解高性能Redis的本质
Redis系列2:数据持久化提高可用性
Redis系列3:高可用之主从架构
Redis系列4:高可用之Sentinel(哨兵模式)
Redis系列5:深入分析Cluster 集群模式
追求性能极致:Redis6.0的多线程模型
追求性能极致:客户端缓存带来的革命
Redis系列8:Bitmap实现亿万级数据计算
Redis系列9:Geo 类型赋能亿级地图位置计算
Redis系列10:HyperLogLog实现海量数据基数统计
Redis系列11:内存淘汰策略
Redis系列12:Redis 的事务机制
Redis系列13:分布式锁实现
Redis系列14:使用List实现消息队列
1 介绍
我们上一篇介绍了如何使用List实现消息队列么,但是我们也看到很多局限性,如下:
- 不支持消息确认机制,没有很好的ACK应答
- 不支持消息回溯,无法排查问题和做消息分析
- List遵循FIFO机制,所以存在消息堆积的风险。
- 查询效率低,作为线性结构,List中定位一个数据需要进行遍历,O(N)的时间复杂度。
- 不存在消费组(Consumer Group)的概念,无法进行分组消费和批量消费
Redis中有三种消息队列模式:
名称 | 简要说明 |
---|---|
List | 不支持消息确认机制(Ack),不支持消息回朔 |
pubSub | 不支持消息确认机制(Ack),不支持消息回朔,不支持消息持久化 |
stream | 支持消息确认机制(Ack),支持消息回朔,支持消息持久化,支持消息阻塞 |
可以看出,作为Redis 5.0 引入的专门为消息队列设计的数据类型,Stream 功能更加健全,更适合做消息队列分发。
Stream 可以包含 0个 到 n个元素的有序队列,并根据ID的大小进行排序。
Stream类型消息队列的具备以下命令特点:
- 可以序列化生成消息ID,方便索引、排序
- 消息可回朔
- 支持Consumer Groups 消费组:多消费者消息争抢,加快消费速度
- 可以阻塞读取消息和非阻塞读取消息
- 没有消息漏读风险
- 有ACK消息确认机制,保证消息至少被消费一次
- 支持多播模式:可以让队列从逻辑上分组进行隔离消费
这些特性,基本达到了一个消息中间件的基本能力,比如:
- 类似 Kafka 的 Consumer Groups 的概念,它也具备了消费组的能力。
- 类似 Rocket MQ的持久化能力,以及高可用的文件存储机制,它也具备了消息的持久化和主从复制机制,可以记录访问位置,方便后续其他时间段继续访问,避免数据丢失。
详细的stream操作见官网文档:https://redis.io/docs/data-types/streams-tutorial/
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:通过该命令,消费者将处理的完的消息标记为已正确完成。
3.1 写入队列数据
咱们先做一下数据准备,创建队列,并往里面写入一些数据,如下:
> 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"
3.2 创建消费者群组
这个的做法就是在队列中创建消费者组,然后指定消费的位置。
语法如下:
# 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
3.3 读取消费组信息
消费队列消息的语法如下:
# 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 的数据。
多个消费者可以达到流量分摊的目的,为大业务流量的场景做负载和分流。如下图,多个消费者相对平均的进行消息消费。
3.4 XPENDING 检查已读取但未ACK的数据
有时候会出现这种情况,就是消费者组或者消费者发生了故障,甚至整个消费者都故障重启了,那么如何避免消息丢失呢,那就是将读取到的但是还没消费的数据进行暂存。
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.5 消息消费完成之后确认(ACK)
正如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队列能力
4.1 添加maven依赖 和 配置基本连接
# maven信息
org.redisson
redisson-spring-boot-starter
3.16.8
# 基本配置
spring:
application:
name: redission_test
redis:
host: x.x.x.x
port: 6379
ssl: false
password: xxxx.xxxx
4.2 Java程序实现
@Slf4j
@Service
public class StreamQueueService {
@Autowired
private RedissonClient redissonClient;
/**
* 生产消息内容
*
* @param msg
* @return
*/
@Override
public void produceMsg(String msg) {
RStream stream = redissonClient.getStream("stream_user");
stream.add("user_id", "1");
stream.add("user_name", "brand");
stream.add("age", "18");
}
/**
* 消费消息内容
*/
@Override
public void consumeMessage() {
// 根据队列名称获取消息队列
RStream stream = redissonClient.getStream("stream_user");
// 创建消费者小组
stream.createGroup("consumer_group1", StreamMessageId.ALL);
// 消费者读取消息
Map> msgs
= stream.readGroup("consumer_group1", "consumer1");
for (Map.Entry> entry : msgs.entrySet()) {
Map msg = entry.getValue();
log.info(msg);
// todo:处理消息的业务逻辑代码
stream.ack("consumer_group1", entry.getKey());
}
}
}
5 总结
相对List,Stream的能力有比较大的提升:
- 支持消息确认机制(ACK应答确认)
- 支持消息回溯,方便排查问题和做消息分析
- 存在消费组(Consumer Group)的概念,可以进行分组消费和批量消费,可以负载多个消费实例
服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net
机房租用,北京机房租用,IDC机房托管, http://www.fwqtg.net
目录 序言 一 、kettle这壶能装些啥 二、Access输入 2.1 准备Acess数据库和表 2.2 新建一个转换并设置 2.3 启动转换预览数据 三、CSV文件输入 3.1 准备csv文件,并将csv输入控件拖入工作区 3.2 csv输入控件设置 3.…