redis队列的使用

快来打我* 2022-10-24 15:26 181阅读 0赞

# 一、redisTemplate对于set集合的使用 #

## 1、需求背景 ##

在做号码下单的时候，我需要去调运营商的接口，查询号码、去对客户的身份信息校验、然后去下预约单、号码锁定、正式单...

首先，号码锁定需要预约单返回的订单号，正式单需要号码锁定返回的seq与预约单返回的订单号

## 2、解决方案 ##

接口的调用顺序   这里，我可以使用一个线程编排去处理这个事情

首先，号码查询与客户信息校验可以同步去进行。

然后根据上一步的结果，去进行预约单

根据预约单结果去号码锁定

根据预约单结果与号码锁定结果去进行正式单

看一下代码

void testCompletableFuture(){
    
            //1 查询订单信息
            CompletableFuture<Object> numberInfos = CompletableFuture.supplyAsync(() -> {
                System.err.println("查询订单信息");
                return new Object();
            });
            //1 校验用户信息
            CompletableFuture<Object> checkUserInfo = CompletableFuture.supplyAsync(() -> {
                System.err.println("校验用户信息");
                return new Object();
            });
    
            //2 预约单  combone将第一步的两个结果都拿到，然后再去执行
            CompletableFuture<Object> preOrder = numberInfos.thenCombine(checkUserInfo, (numberInfo, checkUser) -> {
                System.err.println("下预约单");
                return new Object();
            });
    
            //3 号码预占
            CompletableFuture<Object> lockNumber = preOrder.thenApply(result -> {
                System.err.println("号码预占");
                return new Object();
            });
    
            //4 正式单
            CompletableFuture<Object> order = preOrder.thenCombine(lockNumber, (pre, lock) -> {
                System.err.println("正式单");
                return new Object();
            });
    
            //等待任务完成~~~
            CompletableFuture<String> handle = CompletableFuture.allOf(order).handle((result, exception) -> {
                if (exception != null) {
                    //异常处理
                    return "fail";
                } else {
                    return "success";
                }
            });
            
            try {
                //获取结果~~
                String result = handle.get();
            } catch (Exception e) {
                e.printStackTrace();
            } 
    
        }

## 3、其他问题 ##

但是往往结果并不是想象的那么顺利。有时候，中间返回异常，因为某些号码还需要更多信息，那么就得重新填写扩展信息，重新请求。

而这部分数据肯定就会在我们拿到某一步的结果的时候就得知。  那么这个号码，我会将其保存在redis中，那么下一次对号码下单的话，我就可以从redis查看是否有这个号码，有的话，返回错误，告知就让其填扩展信息，没有就填基本信息就可以了。

并且在每一个号码成功下单完成之后，需要从redis中删除这个号码。因为你不知道redis中是否有这个号码，所以删除操作一定得进行

**将号码添加在redis的set中**

redisTemplate.opsForSet().add(redisKey, number);

在下单成功之后，将其从redis中删除

redisTemplate.opsForSet().remove(redisKey, number);

到了这里基本就结束了。

## 4、代码兼容 ##

但是我的项目要兼容以前的下单流程，而不可能说你直接新建一个流程，按照你的异步编排去走。而以前的流程是是直接一个请求过来，然后依次去调第三方的信息校验、预约单、锁号码、正式单~~~。那么我们上面所有的异步都没办法用到。

所以每一次执行的结果怎么在下一次去拿到？  放在数据库，没问题  。 代价有点大。  放在redis，也没问题。  但是redis是一个基于内存的高性能的缓存数据库，我们需要在里面存频繁需要查询的，并且不是经常更改的数据，或者热点数据。

而我们每一次的执行结果，显然不符合这个条件。使用redis就是大材小用了。  而有一个更加适合的东西去用-----ThreadLocal。

上面也说了，我们是一个请求，需要拿到每一次的结果，那么ThreadLocal就是为这个而出现的。它的线程安全性，并且线程执行完之后，保存的数据也就会销毁，占用的内存就会释放，在这里再合适不过了。所以我就使用了threadLocal来保存每一步执行的结果。

到了这里，需求也基本完美解决了~~~(不要忘记出现异常或者流程结束的时候，去清空这个threadlocal！！！)

# 二、redisTemplate对于zset集合的使用 #

## 1、需求背景 ##

还是下单。在下单完之后，我需要保存这笔订单状态等相关的数据。同时，如果订单状态有变，那么运营商那边会来通知我去更改订单状态。

## 2、问题的出现 ##

下单完成之后，我需要保存这笔订单的seq（我们这边的订单号）,orderNo(第三方订单号),number等，orderChannel(号码下单的运营商)。以后运营商在订单状态更改之后，会给我传过来orderNo,number。所以我要用这两个去查seq与orderchannel，然后新增一条订单状态更改的记录。因为我们这边所有的数据都是跟seq关联的，所以我必须在每一次的运营商回调的时候去查出seq。当然也可以不查seq，在查订单状态的时候，根据seq查到number与orderNo,然后再用这两个去查运营商给我们通知的订单状态。但是系统查订单都是统一的流程，这种特殊情况的处理代码越多，导致后续其他人接手的难度也越大。

在运营商通知到的时候，我去查seq发现并不存在，因为我的订单结果在运营商给我回调之后才返回。所以导致订单状态已经变了，我却还没收到订单成功的通知。这条数据插入收seq为空，而收到订单成功之后，再插入seq，那么表中就存在异常数据。对于异常数据的修复又得去做~~~

所以我使用了zset延时队列，对于运营商的通知订单状态更改并不是需要实时，所以我将信息存起来，等订单结果返回之后再去处理运营商的回调。

## 3、解决方法 ##

入队列  将信息保存在redis中

private static ObjectMapper mapper = Jackson2ObjectMapperBuilder.json().build();
    
    value = mapper.writeValueAsString(param);
                redisTemplate.opsForZSet().add(OrderStatusConfig.ORDER_STATUS_DELAY_KEY, value, System.currentTimeMillis() + 300000);

信息消费并删除

@Scheduled(cron = "0  0/30 * * * ?")
        public void synchronizationOrderStatus(){
            log.info("订单状态信息同步~~~  now：{}", new Date());
            Set set = redisTemplate.opsForZSet().rangeByScore(OrderStatusConfig.ORDER_STATUS_DELAY_KEY, 0, System.currentTimeMillis());
            if(!set.isEmpty()){
                set.stream().forEach(e ->{
                    try{
                        Map map = JSON.parseObject((String)e, Map.class);
                        log.info("消费解析出来的map！：{}", map);
                        orderService.syncHuakaOrder(map);
                        redisTemplate.opsForZSet().remove(OrderStatusConfig.ORDER_STATUS_DELAY_KEY, e);
                    }catch (Exception exception){
                        log.error("订单信息同步异常！原始数据：{}，  error：{}", e, JSONObject.toJSONString(exception));
                    }
                });
            }
        }

**将数据存入zset的时候，给定score为当前时间戳+30分钟，那么取得时候，只取score表示得时间在当前时间之前的数据，取处理这些数据并且删除就ok了**