一、消息队列的简介
在企业的应用中，发送消息方和接收消息方，可以采取同步通信或异步通信。同步通信在实际的应用中效率不高。本文主要介绍异步通信，其中异步通信分为：第一，基于内存的jvm阻塞队列实现异步通信。这种方式面临的问题是：内存空间有限，导致内存泄漏；因为内存的消息泄漏导致敏感信息有泄漏的风险，造成数据安全性问题。第二，基于Redis实现的消息队列，可以解决上述两类问题。
redis在应用中实现消息队列的方式
- Redis 列表list结构是按插入顺序排序的字符串列表。
- Redis 发布/订阅是一种消息传模式，其中发送者（在Redis术语中称为发布者）发送消息，而接收者（订阅者）接收消息。传递消息的通道称为channel。发布订阅 (pub/sub) 可以分发消息，但无法记录历史消息。
- Redis Stream 主要用于消息队列（MQ，Message Queue），Redis 本身是有一个 Redis 发布订阅 (pub/sub) 来实现消息队列的功能，但它有个缺点就是消息无法持久化，如果出现网络断开、Redis 宕机等，消息就会被丢弃。Redis Stream 提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。
MQ的基础概念
生产者：发送消息。消息队列：提供了FIFO的处理机制，具有缓存消息的能力。消费者：接受消息。
Redis 列表list结构实现消息队列
消费者：阻塞获取
brpop l1 30
生产者：
lpush l1 a b c这种方式导致消息丢失，只适合单消费模式。
Redis 发布/订阅是一种消息传模式
消费者：
psubscribe order.* order.queue1
subscribe order.queue1
生产者：
publish order.queue1 msg1
publish order.* msg1这种方式导致数据无法持久化、消息丢失、消息堆积有上限。
基于Stream的消息队列---单消费模式
Redis Streams tutorial | Redis 官网
发送消息命令：
接收消息命令：
基于Stream的消息队列----消费者组（Consumer groups）
创建消费者组命令：
从消费组里读取消息命令：
确认消费组消息命令：
常见的消费组的命令：
# 自动创建消费者组命令
XGROUP CREATE newstream mygroup $ MKSTREAM
# 从消费组读取消息数据
XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream >
XREADGROUP GROUP mygroup Alice STREAMS mystream 0
# 确认消费组已经处理的消息
XACK mystream mygroup 1526569495631-0
# 给定的消费组待挂消息
XPENDING mystream mygroup - + 10详细的Stream的消费组命令参考：https://redis.io/docs/data-types/streams-tutorial/#consumer-groups
二、基于Linux的redis的消息队列的测试
【Stream结构的redis单消费模式】
生产者发送消息：
消费者1接收消息：
消费者2接收消息：
【Stream结构的redis消费组模式】
生产者发送消息：
XGROUP CREATE s1 g1 $ MKSTREAM
消费者1接收消息：
XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >消费者2接收消息：
XREADGROUP GROUP g1 c2 COUNT 1 BLOCK 2000 STREAMS s1 >消费者确认消息：
XACK s1 g1 1683084625732-0三、SpringBoot整合Redis用Stream结构做消息队列
需求场景：1、创建一个stream类型的消息队列（stream.orders）。2、修改下订单的Lua脚本，向消息队列中添加消息。3、项目启动，开启线程任务，尝试获取消息队列中的消息，完成下单。
【1.创建消息队列】
# 创建一个stream类型的消息队列（stream.orders）
XGROUP CREATE stream.orders g1 0 MKSTREAM【2.向消息队列发送消息】
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)【3.从线程池获取线程任务接收消息】
//初始化线程池
private static final ExecutorService seckill_order_executor = Executors.newSingleThreadExecutor();
@PostConstruct
private void init() {
seckill_order_executor.submit(new VoucherOrderHandler());
}
//创建线程任务
private class VoucherOrderHandler implements Runnable {
//从消息队列中获取线程任务
@Override
public void run() {
while (true){
try {
//1.接收消息队列的订单信息 XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds]
// [NOACK] STREAMS key [key ...] id [id ...]
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
Consumer.from("g1", "c1"),
StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
);
//2.判断获取消息是否成功
if (list == null || list.isEmpty()){
//如果失败，进行下一次
continue;
}
//解析消息的订单
MapRecord<String, Object, Object> record = list.get(0);
Map<Object, Object> values = record.getValue();
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
//3.获取消息成功，创建订单
handlerVoucherOrder(voucherOrder);
//4. 确认消息 XACK key group id [id ...]
stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
} catch (Exception e) {
log.error("处理订单异常");
handlerPendinglist();
}
}
}【4. 使用消息队列和业务整合】
private IVoucherOrderService proxy;
//消息队列进行下单
@Override
public Result seckillVoucherOrder(Long voucherId) {
//获取用户
Long userId = UserHolder.getUser().getId();
//订单id
long orderId = redisWork.nextId(SECKILL_ORDER_KEY);
//1.执行Lua脚本
Long result = stringRedisTemplate.execute(
SECKILL_SCRIPT,
Collections.emptyList(),
voucherId.toString(), userId.toString(), String.valueOf(orderId)
);
//2.判断执行的结果是否为0
int r = result.intValue();
if (r != 0){
//2.1 不是0，没有抢购资格
return Result.fail(r == 1 ? "库存不足" : "不允许重复下单");
}
//aop获取代理对象
proxy = (IVoucherOrderService) AopContext.currentProxy();
return Result.ok(orderId);
}【5.测试】
详细的和业务整合代码实现参考链接：https://gitee.com/hfnu_112/springboot_04_dianping
