需要完成功能

借助redis Stream 資料結構實作消息隊列，異步完成訂單建立，其中涉及到了緩存（擊穿，穿透，雪崩），鎖（Redisson），并發處理，異步處理，Lua腳本

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 可以看到組資訊如下：

127.0.0.1:6388> xread count 20 streams s1 0
1) 1) "s1"
   2) 1) 1) "1678852071712-0"
         2) 1) "key1"
            2) "vlaue1"
      2) 1) "1678852073882-0"
         2) 1) "key1"
            2) "vlaue1"
      3) 1) "1678852080406-0"
         2) 1) "key2"
            2) "vlaue2"
      4) 1) "1678852588261-0"
         2) 1) "key-1"
            2) "vlaue2"
      5) 1) "1678852591957-0"
         2) 1) "key-2"
            2) "vlaue2"
      6) 1) "1678852595467-0"
         2) 1) "key-3"
            2) "vlaue2"
      7) 1) "1678852599576-0"
         2) 1) "key-4"
            2) "vlaue2"
      8) 1) "1678852616566-0"
         2) 1) "key-4"
            2) "vlaue2"
      9) 1) "1678852946989-0"
         2) 1) "key-5"
            2) "vlaue2"
127.0.0.1:6388> xinfo groups s1
1)  1) "name"
    2) "g1"
    3) "consumers"
    4) (integer) 1
    5) "pending"
    6) (integer) 5
    7) "last-delivered-id"
    8) "1678852080406-0"
    9) "entries-read"
   10) (integer) 9
   11) "lag"
   12) (integer) 0
127.0.0.1:6388> xreadgroup group g1 lihui  count 1 streams s1 >
1) 1) "s1"
   2) 1) 1) "1678852588261-0"
         2) 1) "key-1"
            2) "vlaue2"
127.0.0.1:6388> xinfo groups s1
1)  1) "name"
    2) "g1"
    3) "consumers"
    4) (integer) 1
    5) "pending"
    6) (integer) 5
    7) "last-delivered-id"
    8) "1678852588261-0"
    9) "entries-read"
   10) (integer) 10
   11) "lag"
   12) (integer) -1
127.0.0.1:6388> xlen s1
(integer) 9
127.0.0.1:6388> xgroup setid s1 g1 1678852073882-0  ENTRIESREAD 2
OK
127.0.0.1:6388> xinfo groups s1
1)  1) "name"
    2) "g1"
    3) "consumers"
    4) (integer) 1
    5) "pending"
    6) (integer) 5
    7) "last-delivered-id"
    8) "1678852073882-0"
    9) "entries-read"
   10) (integer) 2
   11) "lag"
   12) (integer) 7

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) 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

因為在redis 的stream 指令中沒有判斷 consumer group 是否存在，是以需要變通的方法，解決這個問題，因為如果組不存的情況下，進行xadd 會報錯。

建立隊列和消費者組

/**
     * 查詢隊列資訊，如果不存在會出現異常，在異常中建立隊列
     * @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());
            }
        }


    }

判斷隊列是否存在

/**
     * 判斷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;
        }


    }

訂單建立入口函數（放在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<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);

    }

建立線程擷取消息隊列并處理

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);
                }


            }
        }

異常處理在redis stream pending List 中擷取隊列

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);
                    }

                }
        }
    }

訂單處理

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
        }

    }

實作類中的方法

@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);


    }

Java+Redis 通過Lua 完成庫存扣減，建立消息隊列，異步處理消息

需要完成功能

1、讀取庫存資料【Lua】

2、判斷庫存【Lua】

3、扣減庫存【Lua】

4、建立隊列群組【Java】

5、發送隊列消息【Lua】

6、消息讀取并處理【Java】

繼續閱讀

秒懂JVM的三大參數類型，就靠這十個小實驗了初級—中級—進階三個級别的大廠面試真題阿裡雲——Java 實習生/初級美團——Java 中級螞蟻金服——Java 進階基礎篇JVM 篇MySQL 篇Redis 篇

CentOs7 安裝redis4.0 遇到的坑一、CentOs7連接配接網絡二、make檔案的時候出錯三、安裝ruby的redis插件時候報錯最後

docker 搭建Redis 問題解決

如何解決Redis緩存擊穿、雪崩、穿透問題

java 版本的redis-stat不能運作在背景和daemon

redis管理常用指令

django短信驗證碼的後端實作

Redis訂閱了一段時間後訂閱失效了（ redisTemplate.convertAndSend）

2022秋招面試總結（cpp+java+測開）百度測開一面位元組後端一面蝦皮後端一面蝦皮後端二面

資料遷移方法資料遷移原則資料遷移之雙寫方案資料遷移之級聯同步方案

微服務-性能壓測\緩存redis和分布式鎖redisson和SpringCache

Nacos 2.0 更新前後性能對比壓測

Spring資料和Redis

redis叢集資料一緻性_RedisRaft為Redis叢集帶來強大的資料一緻性

supervisor 管理redis 和httpd 環境centos7

Redis簡介一(單機版)發展曆程Redis

Java+Redis 通過Lua 完成庫存扣減，建立消息隊列，異步處理消息

需要完成功能

1、讀取庫存資料 【Lua】

2、判斷庫存 【Lua】

3、扣減庫存 【Lua】

4、建立隊列群組 【Java】

5、發送隊列消息 【Lua】

6、消息讀取并處理 【Java】

繼續閱讀

1、讀取庫存資料【Lua】

2、判斷庫存【Lua】

3、扣減庫存【Lua】

4、建立隊列群組【Java】

5、發送隊列消息【Lua】

6、消息讀取并處理【Java】