当前位置：首页 > news >正文

JAVA-实战8 Redis实战项目—雷神点评（7）Redis消息队列实现异步秒杀

news 2026/5/7 2:27:25

夢の中で描いた絵のようなんだ

Redis消息队列实现异步秒杀

需求:

创建一个Stream类型的消息队列
修改之前的秒杀下单Lua脚本，在认定有抢购资格后，直接向消息队列中添加消息
项目启动时，开启一个线程任务，尝试获取消息队列中的消息，完成下单

创建消息队列

XGROUP CREATE SecondKillMessageQueue SecondKillGroup 0 MKSTREAM

修改Lua脚本

修改Lua脚本如下：

local VoucherId = ARGV[1]
local UserId = ARGV[2]
-- 新增：传递订单Id
local VoucherOrderId = ARGV[3]local StockKey = 'seckill:stock:' .. VoucherId
local OrderKey = 'seckill:order:' .. VoucherIdif(tonumber(redis.call('get',StockKey)) <=0) thenreturn 1
endif(redis.call('sismember',OrderKey,UserId) ==1) thenreturn 2
endredis.call('incrby',StockKey,-1)
redis.call('sadd',OrderKey,UserId)
-- 新增：Redis向消息队列中添加消息，其中键值对的字段名要和数据实体类的属性名一致，从而便于接下来的映射
redis.call('xadd','SecondKillMessageQueue','*','UserId',UserId,'VoucherId',VoucherId,'VoucherOrderId',VoucherOrderId);
return 0;

修改线程任务和消息队列读取

我们不再需要声明成员变量来设置阻塞队列，而是通过stringRedisTemplate.opsForStream().read()方法来直接读取Stream结构的消息队列来获取异步消息

private static final ExecutorService SECONDKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();@PostConstruct
private void init() {SECONDKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}private class VoucherOrderHandler implements Runnable {private final String MessageQueueName = "SecondKillMessageQueue";private final String GroupName = "SecondKillGroup";private final String ConsumerName = "SecondKillConsumer";@Overridepublic void run() {while(true) {try {List<MapRecord<String,Object,Object>> VoucherOrderRecordList = stringRedisTemplate.opsForStream().read(// 三行分别是：设置消费者组名和消费者名，设置单次读取最大量和最长等待时间，设置消息队列名称和获取消息起始IdConsumer.from(GroupName,ConsumerName),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2L)),StreamOffset.create(MessageQueueName, ReadOffset.lastConsumed()));if(VoucherOrderRecordList == null || VoucherOrderRecordList.isEmpty()) {continue;}// 提取消息Id和消息内容（键值对格式）MapRecord<String,Object,Object> VoucherOrderRecord = VoucherOrderRecordList.get(0);RecordId VoucherOrderRecordId = VoucherOrderRecord.getId();Map<Object,Object> VoucherOrderRecordValue = VoucherOrderRecord.getValue();// 设置Map键字段名到Bean属性名的对应关系CopyOptions copyOptions = CopyOptions.create().setFieldMapping(MapUtil.builder("VoucherOrderId", "id")   // Map键 -> Bean属性   .put("UserId", "userId").put("VoucherId", "voucherId").build());VoucherOrderData NewVoucherOrder =  BeanUtil.fillBeanWithMap(VoucherOrderRecordValue, new VoucherOrderData(), copyOptions);// 执行数据库相关操作AddVoucherOrderFromQueueToDatabase(NewVoucherOrder);// 执行完成后确认消息，传递消息队列名称、消费者组名称、消息IDstringRedisTemplate.opsForStream().acknowledge(MessageQueueName,ConsumerName,VoucherOrderRecordId);} catch (Exception e) {log.error("Calc Order Message Failed:",e);// 出现没有被确认的消息，即Pending，进行处理HandlePendingList();}}}// 除PendingList为空处理和最终异常处理不同外，其余均和正常处理相同private void HandlePendingList(){while(true) {try {List<MapRecord<String,Object,Object>> VoucherOrderRecordList = stringRedisTemplate.opsForStream().read(Consumer.from("SecondKillGroup","SecondKillConsumer"),StreamReadOptions.empty().count(1),StreamOffset.create(MessageQueueName, ReadOffset.from("0")));if(VoucherOrderRecordList == null || VoucherOrderRecordList.isEmpty()) {// 失败 表示PendingList无异常break;}MapRecord<String,Object,Object> VoucherOrderRecord = VoucherOrderRecordList.get(0);RecordId VoucherOrderRecordId = VoucherOrderRecord.getId();Map<Object,Object> VoucherOrderRecordValue = VoucherOrderRecord.getValue();CopyOptions copyOptions = CopyOptions.create().setFieldMapping(MapUtil.builder("VoucherOrderId", "id")   // Map键 -> Bean属性.put("UserId", "userId").put("VoucherId", "voucherId").build());VoucherOrderData NewVoucherOrder =  BeanUtil.fillBeanWithMap(VoucherOrderRecordValue, new VoucherOrderData(), copyOptions);AddVoucherOrderFromQueueToDatabase(NewVoucherOrder);stringRedisTemplate.opsForStream().acknowledge(MessageQueueName,"SecondKillConsumer",VoucherOrderRecordId);} catch (Exception e) {log.error("Calc Order PendingList Message Failed:",e);}}}
}

实现过程中，值得注意的点：

stringRedisTemplate.opsForStream().read()方法对应XREAD操作，而获取消息起始Id由ReadOffset.lastConsumed()设置，其源代码执行为：return new ReadOffset(">");，也就是返回>，从下一个未消费的消息开始

MapRecord是SpringDataRedis‌框架中定义的一个Java接口，用于在Java应用程序中表示Redis.Stream中的一条消息记录，其底层数据结构对应Redis.Stream中的Field-ValuePairs（字段-值对）

其接口签名为public interface MapRecord<S, K, V> extends Record<S, Map<K, V>>, Iterable<Map.Entry<K, V>>

S (Stream)：Stream 的名称（即Redis中的Key，例如SecondKillMessageQueue）。
K (Key)：消息中字段的类型（通常为String或Object）。
V (Value)：消息中值的类型（通常为String或Object）。

MapRecord和Redis.Stream的对应关系

MapRecord.getStream()获取StreamKey，也就是Stream名称、Redis键名，例如SecondKillMessageQueue
MapRecord.getId()获取消息唯一标识，也就是消息ID，例如1778089608104-0
MapRecord.getValue()获取消息内容，即XADD发送的键值对

调用BeanUtil.fillBeanWithMap()方法，将Stream的Value（Map类型）映射到下单记录数据实体NewVoucherOrder（VoucherOrderData.class类型）时，需保证前者的字段名和后者的属性名严格匹配，相关手段如下：

①确保字段名严格一致
②可以使用CopyOptions.create().setFieldMapping()方法封装MapBuilder（MapBuilder使用MapUtil.builder(MapKeyName,ClassPropertyName).put(MapKeyName2,ClassPropertyName2)....build()方法来构建）来构建CopyOptions从而配置字段映射
③使用忽略大小写的转换方法BeanUtil.fillBeanWithMapIgnoreCase()