使用Redis的api实现mq生产消息的异步补偿

灰太狼 2023-07-03 07:13 30阅读 0赞

在生产环境中各种应用的开发越来越多的使用到了mq消息队列，很多mq的中间件都实现了消费者消费消息失败的补偿机制，但是mq的生产者发送消息失败是个令人头疼的问题，因为mq并未对其做一些补偿，所以今天就来介绍一个利用Redis实现的补偿机制。  
  使用Redis作为补偿的思路就是利用了Redis中的各种api，如果业务流程中生产者发送消息失败业务流程感知到立刻调用storemsg方法将消息存储到Redis中，然后监听器监听到消息可以进行重试发送。在storemsg方法中利用了Redis的普通String类型将一个随机的uuid作为键msg作为消息进行存储，然后利用redis的list数据结构为msg的key做一个索引，同时利用了Redis的zset数据结构存储了每个消息的重试次数，因为这个次数要每次取出来和10作比较所以利用zset这个数据结构会快一点。然后就是发送消息的实现了，在业务代码中应该建立监听器，或者定时器来执行补偿方法，在补偿方法中尝试发送消息，如果失败则取出其重试次数与10作比较，当大于10的时候就以为着消息无法发送就要移入死信队列，如果发送成功了就要移入历史队列，此处的死信队列和历史队列都是用了list这种数据结构。接下来是代码的具体实现：

import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.Map;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantLock;
    
    import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import org.springframework.data.redis.core.RedisTemplate;
    
    
    public class MrtMqCompensation {
    
    private static Log log = LogFactory.getLog(MrtMqCompensation.class);
    /**
     * 锁的key
     */
    private String key_lock = "RMQ:COMPENSATION:LOCK";
    /**
     * 错误消息的key
     */
    private String key_error_msg = "RMQ:COMPENSATION:ERROR_MSG";
    /**
     * 死信队列的key
     */
    private String key_dead_msg = "RMQ:COMPENSATION:DEAD_MSG";
    /**
     * 消息前缀的key
     */
    private String key_msg = "RMQ:MSG:";
    /**
     * 消息补偿历史的key
     */
    private String key_history = "RMQ:COMPENSATION:HISTORY_";
    /**
     * 定义一个可重入锁
     */
    private ReentrantLock lock = new ReentrantLock();
    
    private RedisTemplate<String, Object> redisTemplate;
    
    // 思路 此处可以选择存储类型，mysql或者redis存储，目前首选redis存储，redis存储时，
    // 需要序列化对象,此处的msg对象是发送失败的消息在真实的生产环境中需要对其做序列化操作
    
    public void storeMsg(Map<String, Object> msg) {
        // 消息的UUID
        String uuid = (String) msg.get("UUID");
        if (null != uuid) {
    
            // 按照redis的说明，key和value必须都是序列化之后的
            // msg中，body字段是序列化的，需要反序列化之后，放入原msg中再放入redisTemplate
            msg.put("BODY", msg.get("BODY"));
            // 放入重试次数
            msg.put("retryCount", 0);
    
            // 消息本体以uuid为key放入redis，未处理的消息 是7天有效期
            String msgKey = key_msg + uuid;
            redisTemplate.opsForValue().set(msgKey, msg, 7, TimeUnit.DAYS);
    
            // uuid放入索引，索引不过期，将所有的key建立一个索引
            redisTemplate.opsForList().leftPush(key_error_msg, msgKey);
    
    
        }
    }
    
    // 此方法是实现补偿的执行类， 以便统一实现补偿，入参为补偿步长
    // 补偿过程中，涉及到三个对象，失败消息list(MQ:COMPENSATION:ERROR_MSG)、
    // 按天的发送历史list("MQ:COMPENSATION:HISTORY_20181113")、消息补偿锁(MQ:COMPENSATION:LOCK)
    // 实现思路，首先试图锁定消息补偿锁，锁定后，获取末尾消息，并发送，发送成功，转移末尾消息到历史中，并获取下一条，
    // 发送失败，停止发送，结束任务；未获取补偿锁的情况下，结束执行，等待下一次调度时获取补偿锁
    // 整个操作中，迁移的一直是消息索引，不是实体，消息实体的有效期是3天，历史队列的有效期也是三天
    
    public void compensation(Integer size) {
    
        if (log.isDebugEnabled()) {
            log.debug("compensation start");
        }
        // 本次执行，对应的历史队列ID
        SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
        String today = df.format(new Date());
        String historyQueueId = key_history + today;
        // 抢锁 setIfAbsent是推荐的抢锁方式 但是此方法没有同时设置过期时间的选项，也就代表着有极限情况，在刚抢到锁之后，就被kill了
        // 需要提前判断
        // 去除历史锁，首先尝试是否有key的存在，尝试如果存在观察是否过期半个小时（经验值），如果过期半个小时，就删掉key
        if (redisTemplate.hasKey(key_lock)) {
            String lastUpdateTime = (String) redisTemplate.opsForValue().get(key_lock);
            if (System.currentTimeMillis() - Long.valueOf(lastUpdateTime) > 1000 * 60 * 30) {
                redisTemplate.delete(key_lock);
            }
        }
    
        try {
            lock.lock();
            // ready为true是抢到锁了。得到失败的队列中的数据数量
            Long allMsgSize = redisTemplate.opsForList().size(key_error_msg);
            if (log.isDebugEnabled()) {
                log.debug("all error msg size:" + allMsgSize);
            }
    
            while (allMsgSize > 0) {
                // 因为没有很好的办法，能够保证应用被kill掉时，消息不丢，所以采用了队尾迁移到队列头部的原子操作，导致必须一个一个消息取出来处理，然后迁移
                // 同样，这样的while+循环是一件很危险的事，目前没有好办法
                // 有可能获取不到步长长度的数据，要做异常处理
                if (size > allMsgSize.intValue()) {
                    size = allMsgSize.intValue();
                }
    
                for (int i = 0; i < size; i++) {
    
                    // 获取最后一个消息的id，index（...,-1）是获取最后一个数据的值
                    String uuid = (String) redisTemplate.opsForList().index(key_error_msg, -1);
    
                    // 获取消息
                    Map msg = (Map) redisTemplate.opsForValue().get(uuid);
    
                    // msg中，body字段是序列化的，需要反序列化之后，放入原msg中再放入redisTemplate
                    //正常情况下这里是将msg中的BODY反序列化再放入msg，所以此处看上去有些多余，因为这是省略了序列化的过程
                    msg.put("BODY", msg.get("BODY"));
    
                    String success = null;
                    //success作为是否发送成功的标志传入
                    //********************此处是具体使用rocketmq发送消息的实现***************************
    
                    //...............................success = ..........
    
                    //*******************************************************
    
                    //此处是发送失败了
                    if (null == success) {
                        // 获取重试次数,使用了redis的zset的api用来存储每个消息的重试次数
                        int retryCount = redisTemplate.opsForZSet().incrementScore("key_retry_count", uuid, 1).intValue();
                        // 处理失败10次以上证明此消息无法发送
                        if (retryCount > 10) {
                            //已经试了十次了实在没有办法发送了就移入死信队列
                            redisTemplate.opsForList().rightPopAndLeftPush(key_error_msg, key_dead_msg);
                            if (log.isDebugEnabled()) {
                                log.debug("移入死信队列:" + uuid);
                            }
                        }
                    } else if (success.equals("SEND_OK")) {
                        // 处理过的消息，有效期改为3天
                        redisTemplate.expire(uuid, 3, TimeUnit.DAYS);
                        //处理成功了就移入历史队列
                        redisTemplate.opsForList().rightPopAndLeftPush(key_error_msg, historyQueueId);
                    }
                }
    
                log.debug("延长锁时间 暂定3分钟 成功！");
                // 延长标志位
                redisTemplate.opsForValue().set(key_lock, System.currentTimeMillis());
                log.debug("延长标志位 成功！");
                //重置队列长度
                allMsgSize = redisTemplate.opsForList().size(key_error_msg);
                log.debug("重置队列长度 成功！");
            }
            // 正常结束业务，刷新历史队列有效期
            redisTemplate.expire(historyQueueId, 3, TimeUnit.DAYS);
            log.debug("正常结束业务，刷新历史队列有效期 成功！");
            //释放锁
            lock.unlock();
            log.debug("结束业务，清理锁 成功！");
        } catch (Exception e) {
            log.error("出现异常", e);
        } finally {
            lock.unlock();
        }
    }

以上的锁使用了单机环境下的java锁，如果使用了集群模式建议采用分布式锁。