多级缓存实现消息投递（短信发送）

蔚落 2021-11-13 22:24 273阅读 0赞

## 多级缓存实现消息投递（短信发送） ##

### 文章背景 ###

接产品需求，要在项目中实现代扣失败后短信的发送，但从这一点来讲比较容易实现，代扣失败直接发短信就是了，但是需求难实现主要是在于场景的复杂性，目前代扣的场景包括，实时代扣、查证、回调三个部分；每天代扣次数不定，目前是两次。

### 要求 ###

支持当天只能发一条失败短信通知，对于一个客户多个绑定卡的情况，只发送给最优卡绑定的手机；

支持后续开放对多次批扣，多次短信发送的扩展需求，可以通过开关实现。

### 设计 ###

###### 方案一：数据入库 代扣失败入库待发送的消息，成功修改数据状态，当最终为失败时，批量短信发送该消息。 ######

###### 方案二：缓存+数据库，缓存中存放待发送的标志，数据库作为相应数据信息发送的来源，采用分页实现。 ######

###### 方案三：缓存，数据存储、数据过滤，全部采用缓存实现。 ######

### 最终方案 ###

由于项目每天多次批扣，还有宽限期等，所以，本次设计完全采用缓存实现，即方案三。

### 代码设计 ###

首先根据现有的业务逻辑，分析代扣出现的几种情况，实时批扣、代扣查询、代扣回调等。

如果想要在原来代码的基础之上做分情况的处理，一是代码耦合性比较高，二是可读性比较差，不方便扩展和问题定位，所有在处理方面采用公共处理的方式。

### 代码实现 ###

#### 缓存工具 ####

public class Redis { 
    
        private volatile static Redis instance;
    
        private static JedisCluster jedisCluster;
    
        private Redis() { 
            jedisCluster = SpringContext.getBean("jedisCluster");
        }
    
        public static Redis getInstance() { 
            if (instance == null) { 
                synchronized (Redis.class) { 
                    if (instance == null) { 
                        instance = new Redis();
                    }
                }
            }
            return instance;
        }
    
    
        public JedisCluster getJedis() { 
            return jedisCluster;
        }
    
        public void close(JedisCluster jedis) { 
    
        }
    
        public String get(String key) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.get(key);
            } finally { 
                close(jedis);
            }
        }
    
        public String set(String key, String value) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.set(key, value);
            } finally { 
                close(jedis);
            }
        }
    
        public String setex(String key, int seconds, String value) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.setex(key, seconds, value);
            } finally { 
                close(jedis);
            }
        }
    
        public Long expire(String key, int seconds) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.expire(key, seconds);
            } finally { 
                close(jedis);
            }
        }
    
        public Long del(String key) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.del(key);
            } finally { 
                close(jedis);
            }
        }
    
        public Long sadd(String key, String... members) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.sadd(key, members);
            } finally { 
                close(jedis);
            }
        }
    
        public Long scard(String key) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.scard(key);
            } finally { 
                close(jedis);
            }
        }
    
        public Set<String> smembers(String key) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.smembers(key);
            } finally { 
                close(jedis);
            }
        }
    
        public Long hset(String key, String field, String value) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.hset(key, field, value);
            } finally { 
                close(jedis);
            }
        }
    
        public String hget(String key, String field) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.hget(key, field);
            } finally { 
                close(jedis);
            }
        }
    
        public Long hdel(String key, String... fields) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.hdel(key, fields);
            } finally { 
                close(jedis);
            }
        }
    
        public Map<String, String> hgetAll(String key) { 
            JedisCluster jedis = getJedis();
            try { 
                return jedis.hgetAll(key);
            } finally { 
                close(jedis);
            }
        }
    
        public String spop(String key) { 
    
            JedisCluster jedis = getJedis();
            try { 
                return jedis.spop(key);
            } finally { 
                close(jedis);
            }
        }
    
        public <T> void setList(String key, List<T> list) { 
            this.setexList(key, 0, list);
        }
    
    
        public <T> void setexList(String key, int second, List<T> list) { 
            JedisCluster jedis = getJedis();
            try { 
                second = 0 == second ? 当天剩余时间 : second;
                jedis.setex(key, second, JSON.toJSONString(list));
            } finally { 
                close(jedis);
            }
        }
    
        public <T> List<T> getList(String key) { 
            JedisCluster jedis = getJedis();
            try { 
                String stringLists = jedis.get(key);
                return  (List<T>) JSON.parse(stringLists);
            } finally { 
                close(jedis);
            }
        }
    
        public void setMap(String key, Map<String,Object> map) { 
            this.setexMap(key, 0, map);
        }
    
        public void setexMap(String key, int second, Map<String,Object> map) { 
            JedisCluster jedis = getJedis();
            try { 
                second = 0 == second ? Helper.getCurrentDate() + Helper.ONE_DAY - Helper.getCurrentTime() : second;
                jedis.setex(key, second, JSON.toJSONString(map));
            } finally { 
                close(jedis);
            }
        }
    
        public Map<String,Object> getMap(String key) { 
            JedisCluster jedis = getJedis();
            try { 
                return  (Map<String,Object>) JSON.parse(jedis.get(key));
            } finally { 
                close(jedis);
            }
        }
    }

#### 设计缓存一 ####

采用Map+List实现，在Map中过滤数据信息，在List中存取需代扣的数据信息，最后遍历List取出所有数据，进行短信发送处理。

#### 数据保存 ####

private List<Object> judgeCustStatusAndSave(CustDetail detail, Map<String, Object> redisMap, List<Object> objectList) { 
        // 如果当前客户缓存为空,RedisKey为一个枚举类
        Map<String, Object> cacheMap = Redis.getInstance().getMap(RedisKey.CUST_SEND_MSG.getKey() + detail.getCustomerId());
        if (null == cacheMap){ 
            installParam(detail, cacheMap);
        }
        // 缓存中是失败，当前数据是成功时更新状态信息
        if (CustDetailStatus.FAILED.getCode() == (Integer) cacheMap.get("status") &&
            CustDetailStatus.SUCCESS.getCode() == detail.getStatus() ){ 
            // 更新状态为成功
            installParam(detail, cacheMap);
            if (null == objectList){ 
                objectList = new ArrayList<>();
                objectList.add(cacheMap);
            }else { 
                AtomicBoolean exits = new AtomicBoolean();
                // 当代扣失败后又存在代扣成功的数据时，删除list缓存中代扣信息，并修改
                objectList.forEach(o -> { 
                    Map<String, Object> objectMap = (Map<String, Object>) o;
                    // 判断缓存中是否存在该客户信息
                    if (detail.getCustomerId().equals(objectMap.get("customerId"))){ 
                        // 缓存中是失败，当前数据是成功时更新状态信息
                        installParam(detail, objectMap);
                        exits.set(true);
                    }
                });
                // 当缓存中不存在时
                if (!exits.get()){ 
                    objectList.add(cacheMap);
                }
            }
            // 将该客户信息放入缓存
            Redis.getInstance().setMap(RedisKey.CUST_SEND_MSG.getKey() + detail.getCustomerId(), cacheMap);
    
        }
        if (null == objectList){ 
            objectList = new ArrayList<>();
            objectList.add(redisMap);
        }else { 
            AtomicBoolean exits = new AtomicBoolean();
            objectList.forEach(o -> { 
                Map<String, Object> objectMap = (Map<String, Object>) o;
                // 判断缓存中是否存在该客户信息
                if (detail.getCustomerId().equals(objectMap.get("customerId"))){ 
                    // 缓存中是失败，当前数据是成功时更新状态信息
                    if (CustDetailStatus.FAILED.getCode() == (Integer) objectMap.get("status") &&
                        CustDetailStatus.SUCCESS.getCode() == detail.getStatus() ){ 
                        installParam(detail, objectMap);
                    }
                    exits.set(true);
                }
            });
            // 缓存中不存在客户信息时添加
            if (!exits.get()){ 
                Map<String, Object> redisMaps = new HashMap<>(4);
                installParam(detail, redisMaps);
                objectList.add(redisMap);
            }
        }
        return objectList;
    }
    
    private void installParam(CustDetail detail, Map<String, Object> redisMap) { 
        redisMap.put("customerId",detail.getCustomerId());
        redisMap.put("phone", detail.getPhone());
        redisMap.put("msgContent", detail.getResultMsg());
        redisMap.put("status", detail.getStatus());
    }

#### 消息推送 ####

@Service
    public class CustMsgSendJob implements SimpleJobProcessor { 
    
        private static final Logger logger = LoggerFactory.getLogger(CustMsgSendJob.class);
    
        @Autowired
        private CustDetailService custDetailService;
    
        @Override
        public void process(JobContext context) throws Exception { 
            logger.info("Start Task：" + this.getClass().getSimpleName() + " -> " + context.getParameter());
            String sendMsgKey = RedisKey.CUST_SEND_MSG.getKey() + Helper.getCurrentDate();
            AtomicInteger count = new AtomicInteger();
            RedisLock lock = new RedisLock((RedisKey.CUST_SEND_MSG.getKey());
                                           lock.wrap(() -> { 
                                               List<Object> msgList = Redis.getInstance().getList(sendMsgKey);
                                               msgList.forEach(msgObj -> { 
                                                   try { 
                                                       Map<String, Object> msgInitMap = (Map<String, Object>) msgObj;
                                                       if (CustDetailStatus.FAILED.getCode() == (Integer) msgInitMap.get("status")) { 
                                                           // 短信发送成功增加条数统计，具体的发送信息的逻辑
                                                           if (msgService.installMsgMapAndSendMsg(msgInitMap)) { 
                                                               count.getAndIncrement();
                                                           }
                                                       }
                                                   } catch (Exception e) { 
                                                       logger.error("代扣失败发送短信处理异常，redisKey->{}, 异常信息->{}", sendMsgKey, e);
                                                   }
                                               });
                                               logger.info("Finish Task：" + this.getClass().getSimpleName() + " -> " + count);
                                           });
                                           logger.info("Finish Task：" + this.getClass().getSimpleName() + " -> " + count);
    
                                           }
                                           }

#### 设计缓存二 ####

由于采用List会在删除元素和最后发送处理的时候遍历List，当数据量大的时候会出现性能问题，所以需要优化缓存处理。

#### 数据保存 ####

public void cacheOfDeal(CustDetail detail) { 
        try { 
            // 拼接customer key
            String stringKey = getCustomerKey(detail,"string");
            String mapKey = getCustomerKey(detail, "map");
            String setKey = getCustomerKey(detail,"set");
            // 判断当前客户代扣状态，进行相应处理
            judgeAndSaveCustInfo(detail, stringKey, mapKey, setKey);
        } catch (Exception e) { 
            logger.error("客户号->{}, 执行代扣缓存待处理短信异常->{}",detail.getCustomerId(), e);
            DingDingUtil.send(true, this.getClass().getSimpleName() + detail.getCustomerId(), "缓存代扣失败待处理短信异常");
        }
    }
    public void judgeStatusAndSaveCustInfo(CustDetail detail, String stringKey, String mapKey, String setKey) { 
        Redis redis = Redis.getInstance();
        String custStatus = redis.get(stringKey);
        if (StringUtils.isEmpty(custStatus)) { 
            // 初次遍历当前客户代扣数据时，添加代扣失败的数据到Map和Set中
            if (CustDetailStatus.FAILED.getCode() == detail.getStatus()) {  // 如果首次失败
                redis.setex(stringKey, getRemainSeconds(),
                            String.valueOf(CustDetailStatus.FAILED.getCode()));
                String cacheParam = installParam(detail);
                redis.hset(mapKey, stringKey, cacheParam);
                redis.sadd(setKey, stringKey);
            }else if (CustDetailStatus.SUCCESS.getCode() == detail.getStatus()){  // 如果首次成功
                redis.setex(stringKey, getRemainSeconds(),
                            String.valueOf(CustDetailStatus.SUCCESS.getCode()));
            }
        }else {   // 缓存不为空
            // 当且仅当缓存中是失败，当前代扣为成功时执行Map删除操作，更新string value
            if (String.valueOf(CustDetailStatus.FAILED.getCode()).equals(custStatus) &&
                CustDetailStatus.SUCCESS.getCode() == detail.getStatus()){ 
                redis.hdel(mapKey, stringKey);
                redis.setex(stringKey, getRemainSeconds(),
                            String.valueOf(CustDetailStatus.SUCCESS.getCode()));
            }
        }
    }
    
    /** * 组装消息发送参数 * @param detail */
    public String installParam (CustDetail detail){ 
        Map<String, Object> map = new HashMap<>();
        map.put("customerId",detail.getCustomerId());
        map.put("phone", detail.getPhone());
        map.put("msgContent", detail.getResultMsg());
        map.put("status", String.valueOf(detail.getStatus()));
        return JSON.toJSONString(map);
    }
    
    /** * 获取当天剩余时间，需手动计算 * @return */
    private int getRemainSeconds() { 
        // TODO
        return 1;
    }
    
    /** * 获取Redis缓存中的key * @param detail * @return */
    private String getCustomerKey(CustDetail detail, String key) { 
    
        String customerKey = RedisKey.CUST_MSG.getKey() + detail.getCustomerId();
        switch (key) { 
            case "string":
                return customerKey;
            case "map":
                return RedisKey.CUST_SEND_MSG.getKey() + "_map";
            case "set":
                return RedisKey.CUST_SEND_MSG.getKey() + "_set";
            default:
                return RedisKey.CUST_SEND_MSG.getKey();
        }
    }

#### 消息推送 ####

// Job处理
    public Result<Integer> querysendMsg() { 
        AtomicInteger integer = new AtomicInteger();
        Redis redis = Redis.getInstance();
        // 获取缓存中的数据信息
        String setKey = RedisKey.CUST_SEND_MSG.getKey() + "_set";
        String mapKey = RedisKey.CUST_SEND_MSG.getKey() + "_map";
        try { 
            RedisLock lock = new RedisLock(RedisKey.CUST_SEND_MSG.getKey() + this.getClass());
            lock.wrap(() -> { 
                String stringKey = redis.spop(setKey);
                // 当 stringKey不为 null 时处理
                while (null != stringKey){ 
                    if (StringUtils.isNotEmpty(stringKey)) { 
                        String StringMap = redis.hget(mapKey, stringKey);
                        // map不为空代表有代扣数据
                        if (StringUtils.isNotEmpty(StringMap)) { 
                            try { 
                                Map<String, Object> retMap = JSON.parse(StringMap, Map.class);
                                // 具体发短信、保存信息的处理逻辑
                                if (installMsgMapAndSendMsg(retMap)) { 
                                    // 发送完成删除map中的缓存
                                    redis.hdel(mapKey, stringKey);
                                    integer.getAndIncrement();
                                }
                            } catch (ParseException e) { 
                                logger.error("客户key->{}, 数据转换异常->{}", stringKey, e);
                            }
                        }
                    }
                    // 进行下次循环
                    stringKey = redis.spop(setKey);
                }
            });
        } catch (Exception e) { 
            logger.error("代扣短信发送异常 mapKey->{}, 数据转换异常->{}", mapKey, e);
        }
        return new Result<>(integer.intValue());
    }

### 比较两种缓存设计 ###

第一种设计比较简单，很容易想象到和理解，但是由于轮训等原因不利于大数据量处理，影响性能；

第二种设计通过增加一种缓存和改变轮训为spop方式，提高处理的效率，设计更为复杂，但是可靠性更强。

### 更多优秀文章 ###

代码已经在[GitHub][]中更新，更多详情可关注[dwyanewede][]。

> JDK动态代理实现原理  
> [https://blog.csdn.net/shang\_xs/article/details/92772437][https_blog.csdn.net_shang_xs_article_details_92772437]  
> java界的小学生  
> [https://blog.csdn.net/shang\_xs][https_blog.csdn.net_shang_xs]

### 公众号推荐 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW5nX3hz_size_16_color_FFFFFF_t_70]

[GitHub]: https://github.com/dwyanewede/spring-boot/tree/master/springboot-rest/src/main/java/com/sxs/web
[dwyanewede]: https://github.com/dwyanewede
[https_blog.csdn.net_shang_xs_article_details_92772437]: https://blog.csdn.net/shang_xs/article/details/92772437
[https_blog.csdn.net_shang_xs]: https://blog.csdn.net/shang_xs
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW5nX3hz_size_16_color_FFFFFF_t_70]: /images/20211113/c5151288270449bd835c69a26b988fa4.png