Java+Redis 通过Lua 完成库存扣减，创建消息队列，异步处理消息

2023-03-24 22:20| 来源: 网络整理| 查看: 265

需要完成功能

借助redis Stream 数据结构实现消息队列，异步完成订单创建，其中涉及到了缓存（击穿，穿透，雪崩），锁（Redisson），并发处理，异步处理，Lua脚本

IDE:IDEA 2022

1、读取库存数据【Lua】 2、判断库存【Lua】3、扣减库存【Lua】4、创建队列和组【Java】5、发送队列消息【Lua】6、消息读取并处理【Java】

在完成功能之前，需要了解一下redis 中有关stream 数据结构相关的命令

XACK：确认消息已经处理，redis 会在PEL(pending entries List )中移除一个或多个消息。一般情况下一个消息被 XREADGROUP 或 XCLAIM之后会被写入PEL。

XADD: 把消息（Entry ,key-value)追加到队列，默认如果队列不存在会创建，除非使用 NOMKSTREAM ,之后可以通过XREAD ,XREANGE 等命令读取或通过XDEL,XTRIM移除消息

XCLAIM,XAUTOCLAIM 改变PEL 中的消息的所有者

XDEL：在队列中移除一个或多个消息（entry)

XGROUP CREATE :在指定的队列中创建一个消费者组，队列key有且仅能有一个，否则重复会提示：-BUSYGROUP 不存在会提示：ERR no such key ，可以通过选项 MKSTREAM 在不存在时创建

XGROUP CREATECONSUMER:在一个给定的队列和消费者组中创建一个消费者，不能重复。此外在任何使用到消费者的命令中，如果不存在则自动创建。如：XREADGROUP

XGROUP DELCONSUMER:在一个给定的队列和消费者组中移除一个消费者在执行此命令之前相关的PEL 要先执行 XCLAIM 或 XACK进行处理，否则将变得 unclaimable.

XGROUP DESTROY:在给定的队列中删除一个消费者组，相关的consumers 和 PEL 都会被删除，所以执行之前要慎重

XGROUP SETID :重新设置指定的消费者组的最后一个处理的消息的ID，通常这个ID初始值是XGROUP CREATE时指定的，每次执行XREDGROUP 也会修改（更新）这个ID，

例如：想让组内的消费者重新处理队列中的所有的entrys时，可以 XGROUP SETID streamkey groupkey 0，在redis 7.0增加了参数 ENTRIESREAD n ，n 为已读数量，此时，xinfo groups streamKey 可以看到组信息如下：

View Code

XINFO [SRTREAM,GROUPS,CONSUMERS] :获取队列，组，消费者的信息

XLEN:返回 stream中entrys的数量

XPENDING :查询通过消费者组读取但未被确认的entrys

XRANGE :返回队列中的某个区间内的entrys

XREVRANGE: 顺序说XRANGE相反的读取队列中的entry

XREAD:从队列内读取一个或多个entry ,支持block

XREADGROUP:在XREAD的基础上在队列与消费者之间增加了组的概念

XSETID:是一个redis 内部命令，用于记录master 与replicate之间的数据同步的最后一个ID 的记录

好了，下面是是代码

Lua 脚本完成相关功能

--[[ 判断优惠券是否充足优惠券 id ARGV[2] ,key KEYS[1] 判断当前用户是否已经下单用户id ARGV[1] 如果库存充足，且用户没有下单则 1、扣库存 2、保存用户下单信息订单key KEYS[2] 订单ID ARGV[3] stream 队列 key KEYS[3] consumer group key KEYS[4] 利用redis 中set 数据类型的不重复性，进行重复下单信息的记录 --]] local sec_kc=redis.call('get',KEYS[1]) --if(not sec_kc) then -- redis.call('set',KEYS[1],2000) -- sec_kc=2000 --end if(not sec_kc or sec_kc=='{}') then --库存不足 return 3 end if( tonumber(sec_kc)0)//队列与组都已经存在 { //System.out.println(streamKey + "队列" + streamKey + "和组" + groupKey + "都已经存在"); } else { log.debug(ex.getMessage()); } } }

判断队列是否存在

/** * 判断consumer group 是否存在,如果队列不存在直接返回 false, * 如果队列存在，不论组是否存，都直接创建。 */ private boolean ConsumerGroupExists(String streamKey,String groupKey) { boolean isok=false; try { //默认队列和组不存创建队列 Collection streamCollect=new ArrayList(); streamCollect.add(streamKey); Long c_stream = stringRedisTemplate.countExistingKeys(streamCollect); if(c_stream==0) { isok= false;//队列不存在 return isok; } //队列存在，那么直接创建消费者组， stringRedisTemplate.opsForStream().createGroup(streamKey, ReadOffset.from("0"), groupKey); System.out.println(streamKey + "队列" + streamKey + "存在，组" + groupKey + "创建成功。"); isok=true;//组创建成功 } catch ( Exception ex) { String errx = ex.getMessage().toString(); if(errx.indexOf("BUSYGROUP Consumer Group name already exists")>0)//队列与组都已经存在 { isok=true;//组创建成功 //System.out.println(streamKey + "队列" + streamKey + "和组" + groupKey + "都已经存在"); } else { isok=false;//未知异常 log.debug(ex.getMessage()); } }finally { return isok; } }

订单创建入口函数（放在controller 或 seervice impl）中

public Result secKillVoucher(Long voucherId) { long userID = UserHolder.getUser().getId(); String vouchStockKey = RedisConstants.SECKILL_VOUCHERSTOCK_CACHE_KEY + voucherId; String userid_s = String.valueOf(userID); String vouchOrderKey = RedisConstants.SECKILL_ORDER_CACHE_KEY + voucherId; String streamKey = RedisConstants.REDIS_STREAM_QUEUE_ORDER ; String groupKey = RedisConstants.REDIS_STREAM_GROUP_ORDER; List keys = new ArrayList(); keys.add(vouchStockKey);//库存key KEYS[1] keys.add(vouchOrderKey);//订单key KEYS[2] keys.add(streamKey);//队列key KEYS[3] keys.add(groupKey);//消费组key KEYS[4] //check sec_voucher_stock // 1 查询优惠券 int count = seckillVoucherService.getRedisStock(voucherId); if (count 0) { log.error("优惠券仅限每人一个 count："+count); return; } //5扣减库存 boolean success=seckillVoucherService.update() .setSql("stock=stock-1") .eq("voucher_id", voucher.getVoucherId()).gt("stock",0) .update(); if(!success) { log.error("扣减券库存失败Obj: voucher.getVoucherId():"+ voucher.getVoucherId()); return; } System.out.println("voucher saved"); boolean ds = save(voucher); }

【本文地址】

公司简介

联系我们