6

Redis之消息队列实现

 1 year ago
source link: https://blog.51cto.com/u_15559794/6100323
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.

Redis之消息队列实现

精选 原创
  • ​​秒杀场景​​
  • ​​采用消息队列实现​​
  • ​​List实现消息队列​​
  • ​​PubSub(发布订阅)实现消息队列​​
  • ​​基于Stream实现消息队列​​
  • ​​消费者组​​
  • ​​实践​​
  • ​​总结​​

秒杀问题是非常重要且比较难实现的,如果不进行架构的优化的话,直接访问会给业务系统造成很大的压力…

  • 场景一:双十一出现的秒杀场景,比如部分商品开展活动,特价11.11,但是库存一般不多,甚至只有一台。在并发量比较大的情况下,如果我们不进行一些优化的话,很容易出现线程安全问题,而且可能给系统带来太大压力,导致宕机。

如果采用以下的秒杀架构图:

Redis之消息队列实现_Redis

我们可以采用Redis来保证一人一单,我们可以将判断秒杀库存和校验一人一单放到Redis层面来完成,采用异步下单来进行秒杀。

这里我们就需要使用到全面分布式锁中生成的全局ID生成器,因为订单id是需要满足自增、唯一性的条件的,这里也可以采用一些网上的开源方案雪花算法等衍生都可以,但是这样还是会存在问题。什么问题呢?

下单操作的原子性?同步队列的实现?

  • 下单代码是需要校验一人一单和库存等条件的,需要使用lua脚本来保证操作的原子性。
  • 采用阻塞队列来存放数据存在内存限制问题和数据安全问题

采用消息队列实现

消息队列(Message Queue),也叫存放消息的队列,最简单的消息队列模型包括:

  • 消息队列:存储和管理消息,也叫做消息代理。
  • 生产者:发送消息到消息队列
  • 消费者:从消息队列获取消息并处理消息
Redis之消息队列实现_消息队列_02

在Redis中也提供了三种方式实现消息队列:

  • List结构:基于List模拟消息队列
  • PubSub:点对点的消息模型
  • Stream:比较完善的消息队列模型
List实现消息队列

在Redis中List的数据结构本身就是一个双向链表,所以很容易就可以模拟出队列的效果。

我们可以通过LPUSH结合RPOP、RPUSH结合LPOP实现,不过当队列中没有消息时RPOP或LPOP操作会返回null,并不会像阻塞队列一样等待消息,因此我们也可以使用BRPOP或者BLPOP来实现阻塞的效果。

优点

  • 优点非常明显,不再受限于JVM内存(阻塞队列消费)
  • 采用Redis的持久化机制,数据安全有保证
  • 可以保证消息的有序性

缺点:

  • 无法避免消息丢失
  • 仅支持单消费
PubSub(发布订阅)实现消息队列

消费者可以订阅一个或多个channel,生产者发送对应channel消息后,所以订阅者都能收到相关消息。

乍一看,好像这个确实比List强多了,使用发布订阅模式,还支持多订阅。

Redis之消息队列实现_消息队列_03

但是缺点也很明显:

  • 不支持数据持久化
  • 无法避免消息丢失
  • 消息堆积有限制,超过时也会出现数据丢失
基于Stream实现消息队列

Stream是Redis5.0以后提出的一种数据类型,是功能比较完善的消息队列

发送消息:

Redis之消息队列实现_消息队列_04

读取消息:

Redis之消息队列实现_redis_05

除去可以实现读取外,还可以设置阻塞时间,从而实现持续监听的效果

Redis之消息队列实现_Redis_06

上面的ID指定为$时,标识读取的是最新的消息,如果有同时两条消息同时到达MQ,下次读取可能会出现漏读的问题

将多个消费者划分到一个组中,监听同一个队列。

Redis之消息队列实现_Redis_07

在Redis的Stream的ID策略中

  • ‘>’: 从下一条未消费的消息开始
  • 其它:根据指定id从pending-list中获取已消费但是未确认的消息

这里的确认是因为前面提到可能会丢失消息,而Stream中的消息在经过消费后,需要进行XACK确认,如果没有进行确认就会把这条消息放到pending-list中。

前面提到如何保证秒杀下单的原子性需要使用到lua脚本,在这里面我把对应的key都写死了

-- 1.参数列表
-- 1.1.优惠券id
local voucherId = ARGV[1]
-- 1.2.用户id
local userId = ARGV[2]
-- 1.3.订单id
local orderId = ARGV[3]

-- 2.数据key
-- 2.1.库存key
local stockKey = 'seckill:stock:' .. voucherId
-- 2.2.订单key
local orderKey = 'seckill:order:' .. voucherId

-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if(tonumber(redis.call('get', stockKey)) <= 0) then
-- 3.2.库存不足,返回1
return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then
-- 3.3.存在,说明是重复下单,返回2
return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.5.下单(保存用户)sadd orderKey userId
redis.call('sadd', orderKey, userId)
-- 3.6.发送消息到队列中, XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0

在消费中,采用ExectorService来创建一个线程池来进行订单的消费

private class VoucherOrderHandler implements Runnable {
String queueName = "stream.orders";
@Override
public void run() {
while (true) {
try {
// 1. 获取消息队列中的信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.order >
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
Consumer.from("g1", "c1"),
StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
StreamOffset.create(queueName, ReadOffset.lastConsumed())
);
// 2. 判断消息获取是否成功
if (list == null || list.isEmpty()) {
// 2.1 获取失败,没有消息,进行下一次循环
continue;
}
// 3. 解析消息中的订单信息
MapRecord<String, Object, Object> record = list.get(0);
Map<Object, Object> values = record.getValue();
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
// 3. 获取成功,可以下单
handleVoucherOrder(voucherOrder);
// 4. ACK确认 XACK stream.orders g1 id
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
} catch (Exception e) {
log.error("处理订单异常");
handlePendingLists();
}
}
}

private void handlePendingLists() {
while (true) {
try {
// 1. 获取pendingList队列中的信息 XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream.order 0
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
Consumer.from("g1", "c1"),
StreamReadOptions.empty().count(1),
StreamOffset.create(queueName, ReadOffset.from("0"))
);
// 2. 判断消息获取是否成功
if (list == null || list.isEmpty()) {
// 2.1 获取失败,说明pending-list没有消息,结束循环
break;
}
// 3. 解析消息中的订单信息
MapRecord<String, Object, Object> record = list.get(0);
Map<Object, Object> values = record.getValue();
VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
// 3. 获取成功,可以下单
handleVoucherOrder(voucherOrder);
// 4. ACK确认 XACK stream.orders g1 id
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
} catch (Exception e) {
log.error("处理pend-list订单异常", e);
try {
Thread.sleep(20);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
}
}

Redis只能满足小项目的需求,更大需求可能需要用到更加高级的消息队列,比如Kafka、RabbitMQ、RocketMQ等等,还有好多学问要学习呢…


About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK