电商商城MQ业务补救方案

忘是亡心i 2021-12-08 20:37 211阅读 0赞

电商商城业务需要发送一些MQ消息，这些消息中部分消息根据业务要求需要进行发送失败的补救。

讨论出补救方案是在需要用到的MQ业务确认出哪些业务需要补救，这些业务中发送kafka消息时进行监听回传发送结果。当发送过程中出现异常导致发送失败时，记录下来保存到Mongo中。然后启一个定时任务执行器，定期从Mongo中根据Topic扫描出那些未处理待补救的记录，重新发送kafka消息，发送成功后标记Mongo中记录。

方案时序图如下：

![商城MQ业务补救方案时序图][MQ]

1、Mongo数据集合设计

<table> 
 <thead> 
  <tr> 
   <th>字段名称</th> 
   <th>字段类型</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>topic</td> 
   <td>String</td> 
   <td>kafka topic</td> 
  </tr> 
  <tr> 
   <td>key</td> 
   <td>String</td> 
   <td>唯一消息id</td> 
  </tr> 
  <tr> 
   <td>data</td> 
   <td>String</td> 
   <td>发送消息</td> 
  </tr> 
  <tr> 
   <td>deal</td> 
   <td>Boolean</td> 
   <td>处理状态，缺省值为false</td> 
  </tr> 
  <tr> 
   <td>send_time</td> 
   <td>String</td> 
   <td>消息发送时间，缺省值为系统当前时间</td> 
  </tr> 
  <tr> 
   <td>repair_method</td> 
   <td>String</td> 
   <td>修复方式：auto（自动 缺省）、manual（手动）</td> 
  </tr> 
  <tr> 
   <td>fail_reason</td> 
   <td>String</td> 
   <td>失败原因，自动生成为“kafka客户端连接超时”</td> 
  </tr> 
  <tr> 
   <td>retry_number</td> 
   <td>Integer</td> 
   <td>重试次数，缺省值为2次</td> 
  </tr> 
  <tr> 
   <td>create_time</td> 
   <td>String</td> 
   <td>创建时间，自动生成为系统当前时间</td> 
  </tr> 
  <tr> 
   <td>update_time</td> 
   <td>String</td> 
   <td>更新时间，自动生成为系统当前时间</td> 
  </tr> 
 </tbody> 
</table>

2、发送服务接口

/**
     * @Project Name:immall-service
     * @Package Name:com.leimingtech.mongo.service
     * @Since JDK 1.8
     */
    package com.leimingtech.mongo.service;
    
    import java.util.Map;
    
    /**
     * @Description: MQ服务
     * @Author 胡松 hoofungson@163.com
     * @Date 2019-04-22 10:22
     * @Version V1.0
     */
    public interface IMqService {
    
        /**
         * 发送消息
         *
         * @param topic KafkaTopic名称
         * @param data  数据
         **/
        void send(String topic, String data);
    
        /**
         * 发送消息
         *
         * @param topic  KafkaTopic名称
         * @param data   数据
         * @param remedy 是否需要补救
         **/
        void send(String topic, String data, Boolean remedy);
    
        /**
         * 发送消息
         *
         * @param topic      KafkaTopic名称
         * @param data       数据
         * @param remedy     是否需要补救
         * @param credential 补救凭证
         **/
        void send(String topic, String data, Boolean remedy, Map<String, Object> credential);
    
        /**
         * 发送消息
         *
         * @param topic  KafkaTopic名称
         * @param key    发送Key值
         * @param data   数据
         * @param remedy 是否需要补救
         **/
        void send(String topic, String key, String data, Boolean remedy);
    
        /**
         * 发送消息
         *
         * @param topic      KafkaTopic名称
         * @param key        发送Key值
         * @param data       数据
         * @param remedy     是否需要补救
         * @param credential 补救凭证
         **/
        void send(String topic, String key, String data, Boolean remedy, Map<String, Object> credential);
    
    }

3、接口实现代码：

/**
     * @Project Name:immall-service
     * @Package Name:com.leimingtech.mongo.service.impl
     * @Since JDK 1.8
     */
    package com.leimingtech.mongo.service.impl;
    
    import cn.hutool.core.date.DateUtil;
    import cn.hutool.core.util.IdUtil;
    import com.alibaba.dubbo.config.annotation.Service;
    import com.leimingtech.mongo.service.IMongoService;
    import com.leimingtech.mongo.service.IMqService;
    import com.leimingtech.mongo.service.util.DefaultListenableFutureCallback;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.MapUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.kafka.support.SendResult;
    import org.springframework.stereotype.Component;
    import org.springframework.util.concurrent.ListenableFuture;
    
    import java.util.Date;
    import java.util.Map;
    import java.util.concurrent.ExecutionException;
    
    /**
     * @Description: MQ服务实现
     * @Author 胡松 hoofungson@163.com
     * @Date 2019-04-22 10:32
     * @Version V1.0
     */
    @Slf4j
    @Component
    @Service(timeout = 1000 * 30, version = "1.0.0", filter = {"requestFilter"})
    public class IMqServiceImpl implements IMqService {
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        @Autowired
        private IMongoService iMongoService;
    
        /**
         * 发送消息
         *
         * @param topic KafkaTopic名称
         * @param data  数据
         **/
        @Override
        public void send(String topic, String data) {
            this.kafkaTemplate.send(topic, data);
        }
    
        /**
         * 发送消息
         *
         * @param topic  KafkaTopic名称
         * @param data   数据
         * @param remedy 是否需要补救
         **/
        @Override
        public void send(String topic, String data, Boolean remedy) {
            if (remedy) {
                String key = IdUtil.fastSimpleUUID();
                this.send(topic, key, data, true);
            } else {
                this.send(topic, data);
            }
        }
    
        /**
         * 发送消息
         *
         * @param topic      KafkaTopic名称
         * @param data       数据
         * @param remedy     是否需要补救
         * @param credential 补救凭证
         **/
        @Override
        public void send(String topic, String data, Boolean remedy, Map<String, Object> credential) {
            if (remedy && MapUtils.isEmpty(credential)) {
                log.error("发送kafka消息需要补救时，补救到Mongo中的数据项不能为空！！");
                return;
            }
            if (remedy) {
                String key = IdUtil.fastSimpleUUID();
                this.send(topic, key, data, true, credential);
            } else {
                this.send(topic, data);
            }
        }
    
        /**
         * 发送消息
         *
         * @param topic  KafkaTopic名称
         * @param key    发送Key值
         * @param data   数据
         * @param remedy 是否需要补救
         **/
        @Override
        public void send(String topic, String key, String data, Boolean remedy) {
            if (remedy) {
                try {
                    ListenableFuture<SendResult<String, String>> listenableFuture = this.kafkaTemplate.send(topic, key, data);
                    listenableFuture.addCallback(new DefaultListenableFutureCallback<String>(listenableFuture.get().getProducerRecord()));
                } catch (ExecutionException e) {
                    log.error("", e);
                } catch (InterruptedException e) {
                    log.error("", e);
                }
            } else {
                this.send(topic, data);
            }
        }
    
        /**
         * 发送消息
         *
         * @param topic      KafkaTopic名称
         * @param key        发送Key值
         * @param data       数据
         * @param remedy     是否需要补救
         * @param credential 补救凭证
         **/
        @Override
        public void send(String topic, String key, String data, Boolean remedy, Map<String, Object> credential) {
            //验证接口
            if (remedy && MapUtils.isEmpty(credential)) {
                log.error("发送kafka消息需要补救时，补救到Mongo中的数据项不能为空！！");
                return;
            }
            if (remedy) {
                try {
                    ListenableFuture<SendResult<String, String>> listenableFuture = this.kafkaTemplate.send(topic, key, data);
                    listenableFuture.addCallback(new DefaultListenableFutureCallback<String>(listenableFuture.get().getProducerRecord()));
                } catch (Exception e) {
                    log.info("正在记录错误信息到Mongo...");
                    credential.put("topic", topic);
                    credential.put("key", key);
                    credential.put("data", data);
                    credential.put("deal", false);
                    if (!credential.containsKey("send_time")) {
                        credential.put("send_time", DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
                    }
                    if (!credential.containsKey("repair_method")) {
                        credential.put("repair_method", "auto");
                    }
                    credential.put("fail_reason", "kafka客户端连接超时");
                    if (!credential.containsKey("retry_number")) {
                        credential.put("retry_number", 2);
                    }
                    credential.put("create_time", DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
                    credential.put("update_time", DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
                    this.iMongoService.saveData("kafka_faild", credential);
                    log.error("kafka发送消息{}到topic {}出现异常：", key, topic, e);
                }
            } else {
                this.send(topic, data);
            }
        }
    }

发生DEMO:

/**
     * @Project Name:immall-service
     * @Package Name:com.leimingtech.mongo.service
     * @Since JDK 1.8
     */
    package com.leimingtech.mongo.service;
    
    import cn.hutool.core.util.IdUtil;
    import cn.hutool.json.JSONUtil;
    import lombok.extern.slf4j.Slf4j;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @Description: MQ服务实现
     * @Author 胡松 hoofungson@163.com
     * @Date 2019-04-22 11:02
     * @Version V1.0
     */
    @Slf4j
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class IMqServiceTest {
    
        @Autowired
        private IMqService iMqService;
    
        /**
         * 发送不需要补救的topic消息
         **/
        @Test
        public void test1() {
            this.iMqService.send("immall.jdvop.product.pool.add.all", "");
            log.info("消息发送成功!!");
        }
    
        /**
         * 发送需要补救的topic消息
         **/
        @Test
        public void test2() {
            Map<String, Object> credential = new HashMap<>();
            credential.put("remark", "初始化商品池");
            Map<String, Object> dataMap = new HashMap<>();
            dataMap.put("skuId", 24521L);
            this.iMqService.send("immall.jdvop.product.pool.add.all", IdUtil.fastSimpleUUID(), JSONUtil.toJsonStr(dataMap), true, credential);
            log.info("消息发送成功!!");
        }
    }

博主地址[——凤松轩博客 - 一个小小文艺范儿的个人][-]

[MQ]: /images/20211209/7efe53bb32f944919fd960f490402b69.png
[-]: https://blog.hoofungson.cn/article/detail/23.html