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

柔情只为你懂 2024-03-16 17:42 16阅读 0赞

在完成功能之前 ，需要了解一下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 可以看到组信息如下：

![7806dbec80501c5216dcffe717ff0e37.gif][]

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 脚本 完成 相关功能

![7bf44a2524add10ab3ae8b36be784ec2.gif][]

--[[
     判断优惠券是否充足  优惠券  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) then
        --库存 不足
        return 1
    end
    --完成一人一单重复检测
    if(redis.call('SISMEMBER',KEYS[2],ARGV[1])==1) then
        --订单重复
        return 2
    end
    --扣减库存
    redis.call('incrby',KEYS[1],-1)
    -- 记录订单与用户之间关系 set 数据
    redis.call('sadd',KEYS[2],ARGV[1])
    --获取stream队列中entry数量
    
    --[=[
    local streamcount=redis.call('xlen',KEYS[3])
    --此功能转到java 中完成 CreateStreamAndGroup
    if(streamcount==0) then--没有stream 需要创建
        -- XGROUP CREATE stream.order group1 0 mkstream
        redis.call('XGROUP','CREATE',KEYS[3],KEYS[4],'0','mkstream')
    
    end
    --]=]
    --判断队列是否存在
    local streamExists=redis.call('exists',KEYS[3])
    if(streamExists==0) then
         return 4--队列不存在
    end
    -- 创建消息entry
    --XADD S1 * KEY12 VALUE12
    redis.call('XADD',KEYS[3],'*','userId',ARGV[1],'voucherId',ARGV[2],'id',ARGV[3])--key 的取值与对象 VoucherOrder 中变量属性对应，分别是，userId,voucherId,id(订单id)
    --满足下单条件
    return 0

![0812f255804634221b9a948b9e8eed44.gif][]

因为在redis 的stream 命令中 没有判断 consumer group 是否存在，所以需要变通的方法，解决这个问题，因为如果组不存的情况下，进行xadd 会报错。

创建队列和消费者组

![aa08807d7d0a0ae8b8246eb31e24d8e6.gif][]

/**
         * 查询队列信息，如果不存在会出现异常，在异常中创建队列
         * @param streamKey
         * @param groupKey
         */
        private void CreateStreamAndGroup(String streamKey,String groupKey)
        {
            try {
                //默认队列和组不存 创建队列
                stringRedisTemplate.opsForStream().createGroup(streamKey, ReadOffset.from("0"), groupKey);
                System.out.println(streamKey + "队列" + streamKey + "和组" + groupKey + "创建成功。");
            }
    
            catch ( Exception ex)
            {
                String errx = ex.getMessage().toString();
    
                if(errx.indexOf("BUSYGROUP Consumer Group name already exists")>0)//队列与组都已经存在
                {
                    //System.out.println(streamKey + "队列" + streamKey + "和组" + groupKey + "都已经存在");
                }
                else
                {
                    log.debug(ex.getMessage());
                }
            }
    
    
        }

![ccc65b592ff1c49f95ed0a83abcd3638.gif][]

判断队列是否存在

![3f68f69b2493a44be1391e87bfa1fd2e.gif][]

/**
         * 判断consumer group 是否存在,如果队列不存在直接返回 false,
         * 如果队列存在，不论组是否存，都直接创建。
         */
        private boolean ConsumerGroupExists(String streamKey,String groupKey)
        {
            boolean isok=false;
            try {
                //默认队列和组不存 创建队列
                Collection<String> 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;
            }
    
    
        }

![a2f1b63d63887d2aa02c543839dcc123.gif][]

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

![8689529195d577f439fda4b160a6349c.gif][]

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<String> 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) {
                return Result.fail("优惠券库存不存在 count：" + count);
    
            }
            long orderid = redisIdWorker.nextId(RedisConstants.ID_ORDER);
            CreateStreamAndGroup(streamKey,groupKey);
    
            //订单ID ARGV[3]
            //优惠券  id ARGV[2]
            //用户id ARGV[1]
            //执行lua 完成优惠券库存，一人一单，检查 并创建用户与优惠券ID之间的关联
            //脚本完成 创建消息队列 （执行此脚本之前确保队列 已经创建
            Long sekResult = stringRedisTemplate.execute(SECKILL_SCRIPT,
                    keys
                    , userid_s,String.valueOf(voucherId),String.valueOf(orderid)
            );
            int ri = sekResult.intValue();
            System.out.println("lua 脚本执行返回值 ："+ri);
            if(ri!=0)
            {
                
                String errtip="优惠券库存不足";
               switch (ri)
               {
                   case 2:
                       errtip="订单重复一人一单";
                           break;
                   case 3:
                       errtip="库存数据不存在";
                       break;
                   case 4:
                       errtip="消息队列不存在";
                       break;
                   case 1:
                       errtip="优惠券库存不足";
                       break;
                   default:
                       errtip="未知错误";
    
               }
    
    
    
              return   Result.fail(errtip);
            }
    　　　　/**
            注意：下面的代理对象要有
     　　　　*/
            proxy=(IVoucherOrderService)AopContext.currentProxy();
    
            return Result.ok(orderid);
    
        }

![07fc6c14e00e597f7e1e79ec347fe48b.gif][]

创建线程获取消息队列并处理

![d864777a09c773511e8523504fd409bc.gif][]

private static final ExecutorService SECKILL_ORDER_EXECUTOR=Executors.newSingleThreadExecutor();
        @PostConstruct
        private void init(){
            //本地阻塞队列的方式
           // SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
            //读取redis消息队列并处理
            SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandlerRedisQueue());
        }
       //从redis 消息队列中获取消息
        private class VoucherOrderHandlerRedisQueue implements Runnable{
    
            @Override
            public void run() {
                while(true){
                    //get orderinfo from blockingqueen
                    try {
                        String streamKey = RedisConstants.REDIS_STREAM_QUEUE_ORDER ;
                        String groupKey = RedisConstants.REDIS_STREAM_GROUP_ORDER;
                        String cs=RedisConstants.REDIS_STREAM_CONSUMER_ORDER;
    
                        //判断组是否存在
                        if (!ConsumerGroupExists(streamKey,groupKey)) {
                           // log.error("异步线程读取redis stream 失败：队列 未创建："+streamKey);
                            Thread.sleep(3000);
                            continue;
                        }
                         //从消息队列中获取消息
                        List<MapRecord<String, Object, Object>> queueList = stringRedisTemplate.opsForStream()
                                .read(Consumer.from( groupKey,cs),
                                StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2))
                                , StreamOffset.create(streamKey,ReadOffset.lastConsumed())
                        );
                        if(queueList==null || queueList.isEmpty() )
                        {
                            //未获取到消息
                            continue;//continue do next
                        }
                        //处理消息
                        MapRecord<String, Object, Object> record = queueList.get(0);
                        Map<Object, Object> vq = record.getValue();
                        VoucherOrder voucherOrder= BeanUtil.fillBeanWithMap(vq,new VoucherOrder(),false);
    
                        //成功获取 创建订单到数据库
                       handlerVoucherOrder(voucherOrder);
                       //确认消息 在PEL中移除
                        stringRedisTemplate.opsForStream().acknowledge(streamKey,groupKey,record.getId());
    
                    } catch (Exception e) {
    
                        HandlePendingList();
                        log.error(e.getMessage().toString());
                        //throw new RuntimeException(e);
                    }
    
    
                }
            }

![01c719e6b8ed8bb4ad11ae793f57c416.gif][]

异常处理 在redis stream pending List 中获取 队列

![9aea0f9429a6c5974f57efba0767608c.gif][]

private void HandlePendingList() {
                String streamKey = RedisConstants.REDIS_STREAM_QUEUE_ORDER ;//stream
                String groupKey = RedisConstants.REDIS_STREAM_GROUP_ORDER;// consumer group
                String cs=RedisConstants.REDIS_STREAM_CONSUMER_ORDER;//consumer
                while (true)
                    try {
                        {
                            //read from  pel
                        List<MapRecord<String, Object, Object>> queueList = stringRedisTemplate.opsForStream()
                                .read(Consumer.from ( groupKey,cs),
                                StreamReadOptions.empty().count(1)
                                , StreamOffset.create(streamKey, ReadOffset.from("0"))
                        );
                        if(queueList==null ||queueList.isEmpty())
                        {
                            //未获取到消息
                            break;//continue do next normal
                        }
                        //处理消息
                        MapRecord<String, Object, Object> record = queueList.get(0);
                        Map<Object, Object> vq = record.getValue();
                        VoucherOrder voucherOrder= BeanUtil.fillBeanWithMap(vq,new VoucherOrder(),false);
    
                        //成功获取
                        handlerVoucherOrder(voucherOrder);
                        //确认消息
                        stringRedisTemplate.opsForStream().acknowledge(streamKey,groupKey,record.getId());
    
                        }
                    } catch (Exception e) {
                        log.debug("消息队列--peding List 处理异常");
                        try {
                            Thread.sleep(50);
                        } catch (InterruptedException ex) {
                            throw new RuntimeException(ex);
                        }
    
                    }
            }
        }

![94733f0ff915d01753598ee0dc8746a9.gif][]

订单处理

![858c8f2e5ed242bc105295e7e2637a84.gif][]

private  IVoucherOrderService proxy;
        private void handlerVoucherOrder(VoucherOrder voucherOrder) {
    
            if(voucherOrder==null || voucherOrder.getUserId()==null)
            {
                log.debug("对象为空 或 属性用户ID 为空。");
            }
            // user ID
            long userID= voucherOrder.getUserId();
    
            RLock lock=redisson.getLock(RedisConstants.LOCK_VOUCHERORDER_KEY+ StrUtil.toString(userID));
            boolean islock = lock.tryLock();//active watch dog
    
            if(!islock)
            {
               log.error("锁创建失败");
            }
    
            try {
                //IVoucherOrderService proxy=(IVoucherOrderService)AopContext.currentProxy();//获取spring 对当前对象的代理
                 proxy.createVoucherOrderByObj(voucherOrder);
    
            } catch (IllegalStateException e) {
                throw new RuntimeException(e);
            }finally {
                //lock.unLock(); //my define simple redis lock
                lock.unlock();//redisson release lock
            }
    
        }

![c31c0b2696c00bb7f6ff2633cb691158.gif][]

实现类中的方法

![9ad28fa9e789e4b2f69e404d035a370f.gif][]

@Transactional//因为方法中 订单的操作 和 库存扣减，所以增加事务支持 为防止数据 不同步
        public void createVoucherOrderByObj(VoucherOrder voucher) {
            // user ID
            if(voucher==null || voucher.getUserId()==null)
            {
                log.debug("对象为空 或 属性用户ID 为空。");
                return;
            }
            long userID=voucher.getUserId();
    
            int count=query().eq("user_id",userID).eq("voucher_id", voucher.getVoucherId()).count();
            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);
    
    
        }

[7806dbec80501c5216dcffe717ff0e37.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/1abc7bfb0c9d467ea57ccba94c95809c.png
[7bf44a2524add10ab3ae8b36be784ec2.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/7c1b99768fd84f97b241e28db11af0da.png
[0812f255804634221b9a948b9e8eed44.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/99bc1a37157d4b7483bf8e017fa1a57e.png
[aa08807d7d0a0ae8b8246eb31e24d8e6.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/2e8574016b074c188fce88859b3382e7.png
[ccc65b592ff1c49f95ed0a83abcd3638.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/9b1f57a280ae44589c7cefcf4ec1a1cf.png
[3f68f69b2493a44be1391e87bfa1fd2e.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/b6b26175fb62479eb4d2d424301554ce.png
[a2f1b63d63887d2aa02c543839dcc123.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/36b5b35f47b747c29d6dfcc38c775055.png
[8689529195d577f439fda4b160a6349c.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/face91db27fc4b1e8669d5bf949f2465.png
[07fc6c14e00e597f7e1e79ec347fe48b.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/6c4d9c3fd71948c5b2858aa431f2c873.png
[d864777a09c773511e8523504fd409bc.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/d8c1dcfa01f24ab282e142c16fc3aebf.png
[01c719e6b8ed8bb4ad11ae793f57c416.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/18f9f2d4c5cc47ca813af2abedba7cf0.png
[9aea0f9429a6c5974f57efba0767608c.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/6bb260be049d428cadae3e4c2bc30308.png
[94733f0ff915d01753598ee0dc8746a9.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/914b82341b0a4c9aa00da22dcbd7019e.png
[858c8f2e5ed242bc105295e7e2637a84.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/7a617b63bd094c5fabc130c424da09da.png
[c31c0b2696c00bb7f6ff2633cb691158.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/0ba3e34dbb284ec28a3a469b900d1c2a.png
[9ad28fa9e789e4b2f69e404d035a370f.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/47a12269ec2e4d2ca94871196e6fd76e.png