黑马点评项目实战（七）——redis消息队列

发表于 更新于 分类于 阅读次数： Waline： 本文字数： 1.7k 阅读时长 ≈ 3 分钟

Redis消息队列

消息队列（Message Queue），存放消息的队列，最简单的模型包括3个角色：

  • 消息队列：存储和管理消息，也成为消息代理（Message Broker）
  • 生产者：发送消息到消息队列
  • 消费者：从消息队列获取消息并处理消息
image-20250217140004950

Redis提供了三种不同的方式来实现消息队列：

  • list结构：基于List结构模拟消息队列
  • PubSub：基本的点对点消息模型
  • Stream：比较完善的消息队列模型

基于List结构模拟消息队列

Redis的list结构是一个双向链表，可以模拟队列效果。不过要注意的是，当队列没有消息时，RPOP或LPOP操作 会返回null，为达到阻塞效果，应该使用BRPOP或BLPOP。

优点：

  • 利用Redis存储，不受限于JVM内存上限。
  • 基于Redis的持久化机制，数据安全性有保证。
  • 可以满足消息有序性。

缺点：

  • 无法避免消息丢失。BRPOP后，如果服务宕机没有处理消息，那么消息就会丢失。
  • 只支持单消费者 。

基于PubSub的消息队列

PubSub是Redis2.0版本引入的消息传递模型。消费者可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关信息。

  • SUBSCRIBE channel [channel]：订阅一个或多个频道
  • PUBLISH channel msg：向一个频道发送消息
  • PSUBSCRIBE pattern[pattern]：订阅与pattern格式匹配的所有频道
image-20250217141845687

优点：

  • 采用发布订阅模型，支持多生产、多消费。

缺点：

  • 不支持数据持久化。
  • 无法避免消息丢失。发布消息如果没人订阅，则消息丢失。
  • 消息堆积有上限，超出时数据丢失。

基于Stream的消息队列

Stream

Stream是Redis5.0引入的一种新数据类型，可以实现一个功能非常完善的消息队列。

image-20250217144341048

image-20250217144518353

在业务中，可以循环的调用XREAD阻塞方式来查询最新消息，从而实现持续监听队列的效果。

1
2
3
4
5
6
7
while (true){
	Object msg = redis.execute("XREAD COUNT 1 BLOCK 2000 STREAMS users $")
	if (msg == null){
		continue;
	}
	handleMessage(msg);
}

但是当我们指定其实ID为$时，代表读取最新的消息，如果我们处理一条消息的过程中，又有超过1条以上的消息到达队列，则下次获取时也只能获取到最新的一条，会出现漏读消息的问题。

XREAD特点：

  • 消息可回溯
  • 一个消息可以被多个消费者读取
  • 可以阻塞读取
  • 有消息漏读的风险

消费者组

image-20250217151059108

image-20250217151246998

image-20250217151510267

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
while (true){
	// 尝试监听队列，使用阻塞模式，最长等待2000ms
	Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >");
	if (msg==null){ // null说明没有消息，继续下一次 
		continue;
	}
	try{
		handleMessage(msg);
	} catch(Exception e){
		while (true){
			Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 STERAMS s1 0");
			if (msg == null){	// null说明没有异常消息，所有消息都已确认，结束循环
				break;
			}
			try{
				//说明有异常消息，再次处理
				handleMessage(msg);
			} catch(Exception e){
				// 再次出现异常，记录日志，继续循环
				continue;
			}
		}
	}
}

XREADGROUP特点：

  • 消息可回溯
  • 可以多消费者争抢消息，加快消费速度
  • 可以阻塞读取
  • 没有消息漏洞的风险
  • 有消息确认机制，保证消息至少被消费一次
image-20250217152914932