添加mq配置增强

bnyer-common/bnyer-common-core/src/main/java/com/bnyer/common/core/constant/ServiceNameConstants.java

@@ -36,4 +36,9 @@ public class ServiceNameConstants
      * 文件服务的serviceid
      */
     public static final String FILE_SERVICE = "bnyer-file";
+
+    /**
+     * pay服务的serviceid
+     */
+    public static final String PAY_SERVICE = "bnyer-pay";
 }

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/config/EnvironmentIsolationConfig.java

@@ -0,0 +1,47 @@
+package com.bnyer.common.rocketmq.config;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description :环境隔离配置：避免测试环境和生产环境可能共用一个RocketMQ环境。如果没有进行处理，
+ * 在测试环境发送的消息就可能被生产环境的消费者消费，生产环境发送的消息也可能被测试环境的消费者消费，从而导致数据混乱的问题。
+ */
+@Configuration
+@RequiredArgsConstructor(onConstructor = @__(@Autowired))
+public class EnvironmentIsolationConfig implements BeanPostProcessor {
+
+    private final RocketEnhanceProperties rocketEnhanceProperties;
+
+    /**
+     * 在装载Bean之前实现参数修改
+     */
+    @Override
+    public Object postProcessBeforeInitialization(@NotNull Object bean, @NotNull String beanName) throws BeansException {
+        if(bean instanceof DefaultRocketMQListenerContainer){
+            DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) bean;
+            if(rocketEnhanceProperties.isEnabledIsolation() && StringUtils.isNotBlank(rocketEnhanceProperties.getEnvironment())){
+                String topic = container.getTopic();
+                String consumerGroup = container.getConsumerGroup();
+                //拼接Topic
+                if (!topic.endsWith(rocketEnhanceProperties.getEnvironment())){
+                    container.setTopic(String.join("-", container.getTopic(),rocketEnhanceProperties.getEnvironment()));
+                }
+                //拼接ConsumerGroup
+                if (!consumerGroup.endsWith(rocketEnhanceProperties.getEnvironment())){
+                    container.setConsumerGroup(String.join("-",container.getConsumerGroup(),rocketEnhanceProperties.getEnvironment()));
+                }
+            }
+            return container;
+        }
+        return bean;
+    }
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/config/RocketEnhanceProperties.java

@@ -0,0 +1,24 @@
+package com.bnyer.common.rocketmq.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description :增强配置类
+ */
+@ConfigurationProperties(prefix = "rocketmq.enhance")
+@Data
+public class RocketEnhanceProperties {
+
+    /**
+     * 启动隔离，启动后会自动在topic上拼接激活的配置文件，达到自动隔离的效果
+     */
+    private boolean enabledIsolation;
+
+    /**
+     * 隔离环境名称，拼接到topic后，如：dev，拼接后：topic-dev，默认空字符串
+     */
+    private String environment;
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/config/RocketMQEnhanceAutoConfiguration.java

@@ -0,0 +1,66 @@
+package com.bnyer.common.rocketmq.config;
+
+import com.bnyer.common.rocketmq.template.RocketMQEnhanceTemplate;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.messaging.converter.MappingJackson2MessageConverter;
+import org.springframework.messaging.converter.MessageConverter;
+
+import java.util.List;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description :
+ */
+@Configuration
+@EnableConfigurationProperties(RocketEnhanceProperties.class)
+public class RocketMQEnhanceAutoConfiguration {
+ 
+    /**
+     * 注入增强的RocketMQEnhanceTemplate
+     */
+    @Bean
+    public RocketMQEnhanceTemplate rocketMQEnhanceTemplate(RocketMQTemplate rocketMQTemplate){
+        return new RocketMQEnhanceTemplate(rocketMQTemplate);
+    }
+ 
+    /**
+     * 解决RocketMQ Jackson不支持Java时间类型配置
+     * 源码参考：{@link org.apache.rocketmq.spring.autoconfigure.MessageConverterConfiguration}
+     */
+    @Bean
+    @Primary
+    public RocketMQMessageConverter enhanceRocketMQMessageConverter(){
+        RocketMQMessageConverter converter = new RocketMQMessageConverter();
+        CompositeMessageConverter compositeMessageConverter = (CompositeMessageConverter) converter.getMessageConverter();
+        List<MessageConverter> messageConverterList = compositeMessageConverter.getConverters();
+        for (MessageConverter messageConverter : messageConverterList) {
+            if(messageConverter instanceof MappingJackson2MessageConverter){
+                MappingJackson2MessageConverter jackson2MessageConverter = (MappingJackson2MessageConverter) messageConverter;
+                ObjectMapper objectMapper = jackson2MessageConverter.getObjectMapper();
+                objectMapper.registerModules(new JavaTimeModule());
+            }
+        }
+        return converter;
+    }
+ 
+ 
+    /**
+     * 环境隔离配置
+     */
+    @Bean
+    @ConditionalOnProperty(name="rocketmq.enhance.enabledIsolation", havingValue="true")
+    public EnvironmentIsolationConfig environmentSetup(RocketEnhanceProperties rocketEnhanceProperties){
+        return new EnvironmentIsolationConfig(rocketEnhanceProperties);
+    }
+ 
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/config/RocketMqAdapter.java

@@ -1,38 +0,0 @@
-package com.bnyer.common.rocketmq.config;
-
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.spring.core.RocketMQTemplate;
-import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.cloud.context.config.annotation.RefreshScope;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * @author :WXC
- * @Date :2023/03/24
- * @description :
- */
-@RefreshScope
-@Configuration
-public class RocketMqAdapter {
-
-    @Autowired
-    private RocketMQMessageConverter rocketMqMessageConverter;
-
-    @Value("${rocketmq.name-server:}")
-    private String nameServer;
-
-    public RocketMQTemplate getTemplateByTopicName(String topic){
-        RocketMQTemplate mqTemplate = new RocketMQTemplate();
-        DefaultMQProducer producer = new DefaultMQProducer(topic);
-        producer.setNamesrvAddr(nameServer);
-        producer.setRetryTimesWhenSendFailed(RocketMqConstant.SYNC_RETRY_FAILED_COUNT);
-        producer.setRetryTimesWhenSendAsyncFailed(RocketMqConstant.ASYNC_RETRY_FAILED_COUNT);
-        producer.setSendMsgTimeout((int) RocketMqConstant.TIMEOUT);
-        mqTemplate.setProducer(producer);
-        mqTemplate.setMessageConverter(rocketMqMessageConverter.getMessageConverter());
-        return mqTemplate;
-    }
-
-}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/config/RocketMqConstant.java

@@ -1,45 +0,0 @@
-package com.bnyer.common.rocketmq.config;
-
-/**
- * @author :WXC
- * @Date :2023/03/24
- * @description :
- */
-public class RocketMqConstant {
-    /**
-     * 默认发送消息超时时间
-     */
-    public static final long TIMEOUT = 3000;
-
-    /**
-     * 延迟队列取消订单时间，实际上30分钟
-     * 按顺序匹配：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h (1-18)
-     */
-    public static final int CANCEL_ORDER_DELAY_LEVEL = 16;
-
-    /**
-     * 发送同步消息失败重试次数，默认2
-     */
-    public static final int SYNC_RETRY_FAILED_COUNT = 2;
-
-    /**
-     * 发送异步消息失败重试次数，默认2
-     */
-    public static final int ASYNC_RETRY_FAILED_COUNT = 2;
-
-    /**
-     * vip订单取消
-     */
-    public static final String VIP_ORDER_CANCEL_TOPIC = "vip-order-cancel-topic";
-
-    /**
-     * vip订单支付成功
-     */
-    public static final String VIP_ORDER_PAY_NOTIFY_TOPIC = "vip-order-pay-notify-topic";
-
-    /**
-     * vip记录创建
-     */
-    public static final String VIP_RECORD_CREATE_TOPIC = "vip-record-create-topic";
-
-}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/constant/RocketMqConstant.java

@@ -0,0 +1,25 @@
+package com.bnyer.common.rocketmq.constant;
+
+/**
+ * @author :WXC
+ * @Date :2023/03/24
+ * @description : mq常量池
+ */
+public class RocketMqConstant {
+
+    /**
+     * 数据来源前缀
+     */
+    public static final String RETRY_PREFIX = "bnyer-";
+
+    /**
+     * 按顺序匹配：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h (1-18)
+     */
+    public static final int FIVE_SECOND = 2;
+
+    /**
+     * 延迟队列取消订单时间，实际上30分钟
+     * 按顺序匹配：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h (1-18)
+     */
+    public static final int THIRTY_MINUTES = 16;
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/constant/RocketMqTopic.java

@@ -0,0 +1,25 @@
+package com.bnyer.common.rocketmq.constant;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description : 消息主题
+ */
+public class RocketMqTopic {
+
+    /**
+     * vip订单取消
+     */
+    public static final String VIP_ORDER_CANCEL_TOPIC = "vip-order-cancel-topic";
+
+    /**
+     * vip订单支付成功
+     */
+    public static final String VIP_ORDER_PAY_NOTIFY_TOPIC = "vip-order-pay-notify-topic";
+
+    /**
+     * vip记录创建
+     */
+    public static final String VIP_RECORD_CREATE_TOPIC = "vip-record-create-topic";
+
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/domain/BaseMessage.java

@@ -0,0 +1,31 @@
+package com.bnyer.common.rocketmq.domain;
+
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description :消息实体，所有消息都需要继承此类
+ */
+@Data
+public class BaseMessage {
+    /**
+     * 业务键，用于RocketMQ控制台查看消费情况
+     */
+    protected String key;
+    /**
+     * 发送消息来源，用于排查问题
+     */
+    protected String source = "";
+    /**
+     * 发送时间
+     */
+    protected LocalDateTime sendTime = LocalDateTime.now();
+    /**
+     * 重试次数，用于判断重试次数，超过重试次数发送异常警告
+     */
+    protected Integer retryTimes = 0;
+
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/domain/img/AddUserVipRecordMessage.java

@@ -0,0 +1,50 @@
+package com.bnyer.common.rocketmq.domain.img;
+
+import com.bnyer.common.rocketmq.domain.BaseMessage;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.util.Date;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description : 添加用户会员记录
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+public class AddUserVipRecordMessage extends BaseMessage {
+
+    @ApiModelProperty(value="订单号")
+    private String orderNo;
+
+    @ApiModelProperty(value="用户id")
+    private Long userId;
+
+    @ApiModelProperty(value="用户手机号")
+    private String phone;
+
+    @ApiModelProperty(value="vip表id")
+    private Long vipId;
+
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    @ApiModelProperty(value="开始时间")
+    private Date startTime;
+
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    @ApiModelProperty(value="到期时间")
+    private Date endTime;
+
+    @ApiModelProperty(value="vip名称")
+    private String vipName;
+
+    @ApiModelProperty(value = "vip类型名称")
+    private String vipTypeName;
+
+    @ApiModelProperty(value = "用户客户端类型：10用户-抖音 20用户-快手 30用户-微信 40艺术家-微信")
+    private Integer userClientType;
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/domain/order/VipOrderCancelMessage.java

@@ -0,0 +1,21 @@
+package com.bnyer.common.rocketmq.domain.order;
+
+import com.bnyer.common.rocketmq.domain.BaseMessage;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description : 会员订单取消
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+public class VipOrderCancelMessage extends BaseMessage {
+    /**
+     * 订单号
+     */
+    private String orderNo;
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/domain/order/VipOrderPayNotifyMessage.java

@@ -0,0 +1,21 @@
+package com.bnyer.common.rocketmq.domain.order;
+
+import com.bnyer.common.rocketmq.domain.BaseMessage;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description : 会员订单支付回调
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+public class VipOrderPayNotifyMessage extends BaseMessage {
+    /**
+     * 订单号
+     */
+    private String orderNo;
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/handle/EnhanceMessageHandler.java

@@ -0,0 +1,151 @@
+package com.bnyer.common.rocketmq.handle;
+
+import com.alibaba.fastjson.JSONObject;
+import com.bnyer.common.rocketmq.constant.RocketMqConstant;
+import com.bnyer.common.rocketmq.domain.BaseMessage;
+import com.bnyer.common.rocketmq.template.RocketMQEnhanceTemplate;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+
+import javax.annotation.Resource;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description : 消息模板抽象类，父类提供公共模板方法
+ */
+@Slf4j
+public abstract class EnhanceMessageHandler<T extends BaseMessage> {
+    /**
+     * 默认重试次数
+     */
+    private static final int MAX_RETRY_TIMES = 3;
+
+    /**
+     * 延时等级
+     */
+    private static final int DELAY_LEVEL = RocketMqConstant.FIVE_SECOND;
+
+    @Resource
+    private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;
+
+    /**
+     * 消息处理
+     *
+     * @param message 待处理消息
+     * @throws Exception 消费异常
+     */
+    protected abstract void handleMessage(T message) throws Exception;
+
+    /**
+     * 超过重试次数消息，需要启用isRetry
+     *
+     * @param message 待处理消息
+     */
+    protected abstract void handleMaxRetriesExceeded(T message);
+
+
+    /**
+     * 是否需要根据业务规则过滤消息，去重逻辑可以在此处处理
+     * @param message 待处理消息
+     * @return true: 本次消息被过滤，false：不过滤
+     */
+    protected abstract boolean filter(T message);
+
+    /**
+     * 是否异常时重复发送
+     *
+     * @return true: 消息重试，false：不重试
+     */
+    protected abstract boolean isRetry();
+
+    /**
+     * 消费异常时是否抛出异常
+     * 返回true，则由rocketmq机制自动重试
+     * false：消费异常(如果没有开启重试则消息会被自动ack)
+     */
+    protected abstract boolean throwException();
+
+    /**
+     * 最大重试次数
+     *
+     * @return 最大重试次数，默认3次
+     */
+    protected int getMaxRetryTimes() {
+        return MAX_RETRY_TIMES;
+    }
+
+    /**
+     * isRetry开启时，重新入队延迟时间
+     * @return -1：立即入队重试
+     */
+    protected int getDelayLevel() {
+        return DELAY_LEVEL;
+    }
+
+    /**
+     * 使用模板模式构建消息消费框架，可自由扩展或删减
+     */
+    public void dispatchMessage(T message) {
+        // 基础日志记录被父类处理了
+        log.info("消费者收到消息[{}]", JSONObject.toJSON(message));
+        if (filter(message)) {
+            log.info("消息id{}不满足消费条件，已过滤。",message.getKey());
+            return;
+        }
+        //超过最大重试次数时调用子类方法处理
+        if (message.getRetryTimes() > getMaxRetryTimes()) {
+            handleMaxRetriesExceeded(message);
+            return;
+        }
+        try {
+            long now = System.currentTimeMillis();
+            handleMessage(message);
+            long costTime = System.currentTimeMillis() - now;
+            log.info("消息{}消费成功，耗时[{}ms]", message.getKey(),costTime);
+        } catch (Exception e) {
+            log.error("消息{}消费异常", message.getKey(),e);
+            // 是捕获异常还是抛出，由子类决定
+            if (throwException()) {
+                //抛出异常，由DefaultMessageListenerConcurrently类处理
+                throw new RuntimeException(e);
+            }
+            //此时如果不开启重试机制，则默认ACK了
+            if (isRetry()) {
+                handleRetry(message);
+            }
+        }
+    }
+
+    protected void handleRetry(T message) {
+        // 获取子类RocketMQMessageListener注解拿到topic和tag
+        RocketMQMessageListener annotation = this.getClass().getAnnotation(RocketMQMessageListener.class);
+        if (annotation == null) {
+            return;
+        }
+        //重新构建消息体
+        String messageSource = message.getSource();
+        if(!messageSource.startsWith(RocketMqConstant.RETRY_PREFIX)){
+            message.setSource(RocketMqConstant.RETRY_PREFIX + messageSource);
+        }
+        message.setRetryTimes(message.getRetryTimes() + 1);
+
+        SendResult sendResult;
+
+        try {
+            // 如果消息发送不成功，则再次重新发送，如果发送异常则抛出由MQ再次处理(异常时不走延迟消息)
+            sendResult = rocketMQEnhanceTemplate.send(annotation.topic(), annotation.selectorExpression(), message, getDelayLevel());
+        } catch (Exception ex) {
+            // 此处捕获之后，相当于此条消息被消息完成然后重新发送新的消息
+            //由生产者直接发送
+            throw new RuntimeException(ex);
+        }
+        // 发送失败的处理就是不进行ACK，由RocketMQ重试
+        if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
+            throw new RuntimeException("重试消息发送失败");
+        }
+
+    }
+}

bnyer-common/bnyer-common-rocketmq/src/main/java/com/bnyer/common/rocketmq/template/RocketMQEnhanceTemplate.java

@@ -0,0 +1,238 @@
+package com.bnyer.common.rocketmq.template;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.bnyer.common.rocketmq.config.RocketEnhanceProperties;
+import com.bnyer.common.rocketmq.constant.RocketMqConstant;
+import com.bnyer.common.rocketmq.domain.BaseMessage;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQHeaders;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+import javax.annotation.Resource;
+
+/**
+ * @author :WXC
+ * @Date :2023/05/17
+ * @description : RocketMq增强请求模板类
+ */
+@Slf4j
+@RequiredArgsConstructor(onConstructor = @__(@Autowired))
+public class RocketMQEnhanceTemplate {
+
+    /**
+     * 延时等级
+     */
+    private static final int DELAY_LEVEL = RocketMqConstant.FIVE_SECOND;
+
+    /**
+     * 默认重试次数
+     */
+    private static final int MAX_RETRY_TIMES = 3;
+
+    private final RocketMQTemplate template;
+
+    @Resource
+    private RocketEnhanceProperties rocketEnhanceProperties;
+
+    public RocketMQTemplate getTemplate() {
+        return template;
+    }
+
+    /**
+     * 根据系统上下文自动构建隔离后的topic
+     * 构建目的地
+     */
+    public String buildDestination(String topic, String tag) {
+        topic = reBuildTopic(topic);
+        return StringUtils.isNotBlank(tag)?topic + ":" + tag:topic;
+    }
+
+    /**
+     * 根据环境重新隔离topic
+     * @param topic 原始topic
+     */
+    private String reBuildTopic(String topic) {
+        if(rocketEnhanceProperties.isEnabledIsolation() && StringUtils.isNoneBlank(rocketEnhanceProperties.getEnvironment())){
+            return topic +"-" + rocketEnhanceProperties.getEnvironment();
+        }
+        return topic;
+    }
+
+    /**
+     * 发送同步消息
+     * @param topic
+     * @param tag
+     * @param message
+     * @return
+     * @param <T>
+     */
+    public <T extends BaseMessage> SendResult send(String topic, String tag, T message) {
+        // 注意分隔符
+        return send(buildDestination(topic,tag), message);
+    }
+
+
+    /**
+     * 发送同步消息
+     * @param destination
+     * @param message
+     * @return
+     * @param <T>
+     */
+    public <T extends BaseMessage> SendResult send(String destination, T message) {
+        // 设置业务键，此处根据公共的参数进行处理
+        // 更多的其它基础业务处理...
+        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
+        SendResult sendResult = template.syncSend(destination, sendMessage);
+        // 此处为了方便查看给日志转了json，根据选择选择日志记录方式，例如ELK采集
+        log.info("[{}]同步消息[{}]发送结果[{}]", destination, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
+        return sendResult;
+    }
+
+    /**
+     * 发送同步延时消息
+     * @param topic
+     * @param tag
+     * @param message
+     * @param delayLevel
+     * @return
+     * @param <T>
+     */
+    public <T extends BaseMessage> SendResult send(String topic, String tag, T message, int delayLevel) {
+        return send(buildDestination(topic,tag), message, delayLevel);
+    }
+
+    /**
+     * 发送同步延时消息
+     * @param destination
+     * @param message
+     * @param delayLevel
+     * @return
+     * @param <T>
+     */
+    public <T extends BaseMessage> SendResult send(String destination, T message, int delayLevel) {
+        Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
+        SendResult sendResult = template.syncSend(destination, sendMessage, 3000, delayLevel);
+        log.info("[{}]延迟等级[{}]消息[{}]发送结果[{}]", destination, delayLevel, JSONObject.toJSON(message), JSONObject.toJSON(sendResult));
+        return sendResult;
+    }
+
+    /**
+     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
+     * （适合对响应时间敏感的业务场景）
+     */
+    public <T extends BaseMessage> void sendAsyncMsg(String topic, String tag, T message) {
+        sendAsyncMsg(buildDestination(topic,tag),message);
+    }
+
+    /**
+     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
+     * （适合对响应时间敏感的业务场景）
+     */
+    public <T extends BaseMessage> void sendAsyncMsg(String destination, T message) {
+        template.asyncSend(destination, MessageBuilder.withPayload(message).build(), new SendCallback() {
+            @Override
+            public void onSuccess(SendResult sendResult) {
+                // 处理消息发送成功逻辑
+                log.info("消息发送成功,destination:{},result:{}",destination, JSON.toJSONString(sendResult));
+            }
+            @Override
+            public void onException(Throwable throwable) {
+                // 处理消息发送失败逻辑
+                log.error("消息发送失败,destination:{},error:{}",destination,throwable.getMessage());
+            }
+        });
+    }
+
+    /**
+     * 发送异步可重试消息（通过线程池执行发送到broker的消息任务，执行完后回调：如果发送失败进行重试）
+     *
+     */
+    public <T extends BaseMessage> void sendAsyncRetryMsg(String topic, String tag, T message) {
+        sendAsyncMsg(buildDestination(topic,tag),message,getDelayLevel());
+    }
+
+    /**
+     * 发送异步延时消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
+     *
+     */
+    public <T extends BaseMessage> void sendAsyncMsg(String topic, String tag, T message, int delayLevel) {
+        sendAsyncMsg(buildDestination(topic,tag),message,delayLevel);
+    }
+
+    /**
+     * 发送异步延时消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
+     * （适合对响应时间敏感的业务场景）
+     */
+    public <T extends BaseMessage> void sendAsyncMsg(String destination, T message, int delayLevel) {
+        template.asyncSend(destination, MessageBuilder.withPayload(message).build(), new SendCallback() {
+            @Override
+            public void onSuccess(SendResult sendResult) {
+                // 处理消息发送成功逻辑
+                log.info("消息发送成功,destination:{},result:{}",destination,JSON.toJSONString(sendResult));
+            }
+            @Override
+            public void onException(Throwable throwable) {
+                log.error("消息发送失败,destination:{},error:{}",destination,throwable.getMessage());
+                // 超过最大重试次数时调用子类方法处理
+                if (message.getRetryTimes() > getMaxRetryTimes()) {
+                    handleMaxRetriesExceeded(message);
+                    return;
+                }
+                //发送失败，重试发送
+                handleRetry(destination,message);
+            }
+            /**
+             * 重试发送异步消息
+             * @param destination
+             * @param message
+             * @param <T>
+             */
+            private <T extends BaseMessage> void handleRetry(String destination,T message) {
+                //重新构建消息体
+                String messageSource = message.getSource();
+                if(!messageSource.startsWith(RocketMqConstant.RETRY_PREFIX)){
+                    message.setSource(RocketMqConstant.RETRY_PREFIX + messageSource);
+                }
+                message.setRetryTimes(message.getRetryTimes() + 1);
+                // 如果消息发送不成功，则再次重新发送
+                sendAsyncMsg(destination, message,getDelayLevel());
+            }
+
+            /**
+             * 超过最大重试次数调用该方法处理
+             * @param message
+             * @param <T>
+             */
+            private <T extends BaseMessage> void handleMaxRetriesExceeded(T message) {
+                log.error("发送消息超过最大重试次数,message:{}",message);
+            }
+        },delayLevel);
+    }
+
+    /**
+     * isRetry开启时，重新入队延迟时间
+     * @return -1：立即入队重试
+     */
+    protected int getDelayLevel() {
+        return DELAY_LEVEL;
+    }
+
+    /**
+     * 最大重试次数
+     *
+     * @return 最大重试次数，默认3次
+     */
+    protected int getMaxRetryTimes() {
+        return MAX_RETRY_TIMES;
+    }
+
+}

bnyer-services/bnyer-img/src/main/java/com/bnyer/img/listener/VipRecordCreateConsumer.java

@@ -1,10 +1,10 @@
 package com.bnyer.img.listener;
 
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
-import com.bnyer.common.core.domain.UserVipRecord;
 import com.bnyer.common.core.dto.AddUserVipRecordDto;
-import com.bnyer.common.rocketmq.config.RocketMqConstant;
+import com.bnyer.common.core.utils.bean.EntityConvertUtil;
+import com.bnyer.common.rocketmq.constant.RocketMqTopic;
+import com.bnyer.common.rocketmq.domain.img.AddUserVipRecordMessage;
+import com.bnyer.common.rocketmq.handle.EnhanceMessageHandler;
 import com.bnyer.img.service.UserVipRecordService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
@@ -19,18 +19,42 @@ import org.springframework.stereotype.Component;
  */
 @Slf4j
 @Component
-@RocketMQMessageListener(topic = RocketMqConstant.VIP_RECORD_CREATE_TOPIC,consumerGroup = RocketMqConstant.VIP_RECORD_CREATE_TOPIC)
-public class VipRecordCreateConsumer implements RocketMQListener<String> {
+@RocketMQMessageListener(topic = RocketMqTopic.VIP_RECORD_CREATE_TOPIC,consumerGroup = RocketMqTopic.VIP_RECORD_CREATE_TOPIC)
+public class VipRecordCreateConsumer extends EnhanceMessageHandler<AddUserVipRecordMessage> implements RocketMQListener<AddUserVipRecordMessage> {
 
     @Autowired
     private UserVipRecordService userVipRecordService;
 
     @Override
-    public void onMessage(String message) {
-        log.info("收到消息：{}", message);
-        JSONObject jsonObject = JSON.parseObject(message);
-        AddUserVipRecordDto addUserVipRecordDto = JSON.toJavaObject(jsonObject, AddUserVipRecordDto.class);
+    public void onMessage(AddUserVipRecordMessage message) {
+        super.dispatchMessage(message);
+        AddUserVipRecordDto addUserVipRecordDto = EntityConvertUtil.copy(message, AddUserVipRecordDto.class);
         //添加用户会员记录
         userVipRecordService.addUserVipRecord(addUserVipRecordDto);
     }
+
+    @Override
+    protected void handleMessage(AddUserVipRecordMessage message) throws Exception {
+
+    }
+
+    @Override
+    protected void handleMaxRetriesExceeded(AddUserVipRecordMessage message) {
+
+    }
+
+    @Override
+    protected boolean filter(AddUserVipRecordMessage message) {
+        return false;
+    }
+
+    @Override
+    protected boolean isRetry() {
+        return true;
+    }
+
+    @Override
+    protected boolean throwException() {
+        return false;
+    }
 }

bnyer-services/bnyer-order/src/main/java/com/bnyer/order/config/RocketMqConfig.java

@@ -1,35 +1,35 @@
-package com.bnyer.order.config;
-
-import com.bnyer.common.rocketmq.config.RocketMqAdapter;
-import com.bnyer.common.rocketmq.config.RocketMqConstant;
-import org.apache.rocketmq.spring.core.RocketMQTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.context.config.annotation.RefreshScope;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Lazy;
-
-/**
- * @author :WXC
- * @Date :2023/03/24
- * @description :
- */
-@RefreshScope
-@Configuration
-public class RocketMqConfig {
-    @Autowired
-    private RocketMqAdapter rocketMqAdapter;
-
-    @Lazy
-    @Bean(destroyMethod = "destroy")
-    public RocketMQTemplate orderCancelMqTemplate() {
-        return rocketMqAdapter.getTemplateByTopicName(RocketMqConstant.VIP_ORDER_CANCEL_TOPIC);
-    }
-
-    @Lazy
-    @Bean(destroyMethod = "destroy")
-    public RocketMQTemplate vipRecordMqTemplate() {
-        return rocketMqAdapter.getTemplateByTopicName(RocketMqConstant.VIP_RECORD_CREATE_TOPIC);
-    }
-
-}
+//package com.bnyer.order.config;
+//
+//import com.bnyer.common.rocketmq.adapter.RocketMqAdapter;
+//import com.bnyer.common.rocketmq.constant.RocketMqTopic;
+//import com.bnyer.common.rocketmq.template.RocketMQEnhanceTemplate;
+//import org.springframework.beans.factory.annotation.Autowired;
+//import org.springframework.cloud.context.config.annotation.RefreshScope;
+//import org.springframework.context.annotation.Bean;
+//import org.springframework.context.annotation.Configuration;
+//import org.springframework.context.annotation.Lazy;
+//
+///**
+// * @author :WXC
+// * @Date :2023/03/24
+// * @description :
+// */
+//@RefreshScope
+//@Configuration
+//public class RocketMqConfig {
+//    @Autowired
+//    private RocketMqAdapter rocketMqAdapter;
+//
+//    @Lazy
+//    @Bean(destroyMethod = "destroy")
+//    public RocketMQEnhanceTemplate vipOrderCancelMqTemplate() {
+//        return rocketMqAdapter.getTemplateByTopicName(RocketMqTopic.VIP_ORDER_CANCEL_TOPIC);
+//    }
+//
+//    @Lazy
+//    @Bean(destroyMethod = "destroy")
+//    public RocketMQEnhanceTemplate vipRecordCreateMqTemplate() {
+//        return rocketMqAdapter.getTemplateByTopicName(RocketMqTopic.VIP_RECORD_CREATE_TOPIC);
+//    }
+//
+//}

bnyer-services/bnyer-order/src/main/java/com/bnyer/order/listener/vip/VipOrderCancelConsumer.java

@@ -1,6 +1,8 @@
 package com.bnyer.order.listener.vip;
 
-import com.bnyer.common.rocketmq.config.RocketMqConstant;
+import com.bnyer.common.rocketmq.constant.RocketMqTopic;
+import com.bnyer.common.rocketmq.domain.order.VipOrderCancelMessage;
+import com.bnyer.common.rocketmq.handle.EnhanceMessageHandler;
 import com.bnyer.order.bean.dto.CancelVipOrderDto;
 import com.bnyer.order.service.VipOrderService;
 import com.google.common.collect.Lists;
@@ -17,19 +19,44 @@ import org.springframework.stereotype.Component;
  */
 @Slf4j
 @Component
-@RocketMQMessageListener(topic = RocketMqConstant.VIP_ORDER_CANCEL_TOPIC,consumerGroup = RocketMqConstant.VIP_ORDER_CANCEL_TOPIC)
-public class VipOrderCancelConsumer implements RocketMQListener<String> {
+@RocketMQMessageListener(topic = RocketMqTopic.VIP_ORDER_CANCEL_TOPIC,consumerGroup = RocketMqTopic.VIP_ORDER_CANCEL_TOPIC)
+public class VipOrderCancelConsumer extends EnhanceMessageHandler<VipOrderCancelMessage> implements RocketMQListener<VipOrderCancelMessage> {
 
     @Autowired
     private VipOrderService vipOrderService;
 
     @Override
-    public void onMessage(String orderNo) {
-        log.info("收到消息：{}", orderNo);
+    public void onMessage(VipOrderCancelMessage message) {
+        super.dispatchMessage(message);
         // 如果订单未支付的话，将订单设为取消状态
         CancelVipOrderDto cancelVipOrderDto = new CancelVipOrderDto();
         cancelVipOrderDto.setOrderStatus(3);
-        cancelVipOrderDto.setOrderNos(Lists.newArrayList(orderNo));
+        cancelVipOrderDto.setOrderNos(Lists.newArrayList(message.getOrderNo()));
         vipOrderService.cancelVipOrder(cancelVipOrderDto);
     }
+
+    @Override
+    protected void handleMessage(VipOrderCancelMessage message) throws Exception {
+
+    }
+
+    @Override
+    protected void handleMaxRetriesExceeded(VipOrderCancelMessage message) {
+        log.error("消息消费失败，可扩展执行后续处理");
+    }
+
+    @Override
+    protected boolean filter(VipOrderCancelMessage message) {
+        return false;
+    }
+
+    @Override
+    protected boolean isRetry() {
+        return true;
+    }
+
+    @Override
+    protected boolean throwException() {
+        return false;
+    }
 }

bnyer-services/bnyer-order/src/main/java/com/bnyer/order/listener/vip/VipOrderPayNotifyConsumer.java

@@ -1,18 +1,19 @@
 package com.bnyer.order.listener.vip;
 
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.bnyer.common.core.constant.ServiceNameConstants;
 import com.bnyer.common.core.domain.VipOrder;
-import com.bnyer.common.rocketmq.config.RocketMqConstant;
+import com.bnyer.common.core.utils.uuid.IdUtils;
+import com.bnyer.common.rocketmq.constant.RocketMqTopic;
+import com.bnyer.common.rocketmq.domain.img.AddUserVipRecordMessage;
+import com.bnyer.common.rocketmq.domain.order.VipOrderPayNotifyMessage;
+import com.bnyer.common.rocketmq.handle.EnhanceMessageHandler;
+import com.bnyer.common.rocketmq.template.RocketMQEnhanceTemplate;
 import com.bnyer.order.mapper.VipOrderMapper;
 import com.bnyer.order.service.VipOrderService;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.rocketmq.client.producer.SendCallback;
-import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
-import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
@@ -25,8 +26,8 @@ import java.util.Objects;
  */
 @Slf4j
 @Component
-@RocketMQMessageListener(topic = RocketMqConstant.VIP_ORDER_PAY_NOTIFY_TOPIC,consumerGroup = RocketMqConstant.VIP_ORDER_PAY_NOTIFY_TOPIC)
-public class VipOrderPayNotifyConsumer implements RocketMQListener<String> {
+@RocketMQMessageListener(topic = RocketMqTopic.VIP_ORDER_PAY_NOTIFY_TOPIC,consumerGroup = RocketMqTopic.VIP_ORDER_PAY_NOTIFY_TOPIC)
+public class VipOrderPayNotifyConsumer extends EnhanceMessageHandler<VipOrderPayNotifyMessage> implements RocketMQListener<VipOrderPayNotifyMessage> {
 
     @Resource
     private VipOrderMapper vipOrderMapper;
@@ -35,14 +36,13 @@ public class VipOrderPayNotifyConsumer implements RocketMQListener<String> {
     private VipOrderService vipOrderService;
 
     @Resource
-    private RocketMQTemplate vipRecordMqTemplate;
+    private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;
 
     @Override
-    public void onMessage(String message) {
-        log.info("收到消息：{}", message);
+    public void onMessage(VipOrderPayNotifyMessage message) {
+        super.dispatchMessage(message);
         //修改订单并添加会员记录
-        JSONObject orderObj = JSON.parseObject(message);
-        String orderNo = orderObj.getString("orderNo");
+        String orderNo = message.getOrderNo();
         VipOrder vipOrder = vipOrderMapper.selectOne(new LambdaQueryWrapper<VipOrder>().eq(VipOrder::getOrderNo, orderNo));
         if (Objects.isNull(vipOrder)){
             log.error("订单不存在，订单号:{}",orderNo);
@@ -51,33 +51,12 @@ public class VipOrderPayNotifyConsumer implements RocketMQListener<String> {
         //修改订单表状态为已支付
         vipOrderService.updateByToPaySuccess(vipOrder);
         //发消息，添加用户会员记录
-        String msg = buildVipRecordMsg(vipOrder);
-        sendMsg(RocketMqConstant.VIP_RECORD_CREATE_TOPIC, msg);
-//        SendStatus sendStatus = vipRecordMqTemplate.syncSend(RocketMqConstant.VIP_RECORD_CREATE_TOPIC, new GenericMessage<>(msg)).getSendStatus();
-//        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
-//            // 消息发不出去就抛异常
-//            throw new ServiceException(ResponseEnum.SERVER_ERROR);
-//        }
+        AddUserVipRecordMessage addUserVipRecordMessage = buildVipRecordMsg(vipOrder);
+        rocketMQEnhanceTemplate.sendAsyncRetryMsg(RocketMqTopic.VIP_RECORD_CREATE_TOPIC,null, addUserVipRecordMessage);
+
         // TODO: 2023/05/17 发消息添加其他奖励
-    }
 
-    /**
-     * 发送消息
-     * @param topic
-     * @param msg
-     * @param <T>
-     */
-    public <T> void sendMsg(String topic, T msg) {
-        vipRecordMqTemplate.asyncSend(topic, msg,new SendCallback() {
-            @Override
-            public void onSuccess(SendResult sendResult) {
-                log.info("topic:{}消息发送成功,result:{}",topic,JSON.toJSONString(sendResult));
-            }
-            @Override
-            public void onException(Throwable e) {
-                log.info("topic:{}消息发送失败,error:{}",topic,e.getMessage());
-            }
-        });
+
     }
 
     /**
@@ -85,18 +64,44 @@ public class VipOrderPayNotifyConsumer implements RocketMQListener<String> {
      * @param vipOrder
      * @return
      */
-    private String buildVipRecordMsg(VipOrder vipOrder) {
-        JSONObject jsonObject = new JSONObject();
-        jsonObject.put("orderNo",vipOrder.getOrderNo());
-        jsonObject.put("startTime",vipOrder.getStartTime());
-        jsonObject.put("endTime",vipOrder.getEndTime());
-        jsonObject.put("vipId",vipOrder.getVipId());
-        jsonObject.put("vipName",vipOrder.getVipName());
-        jsonObject.put("vipTypeName",vipOrder.getVipTypeName());
-        jsonObject.put("userClientType",vipOrder.getUserClientType());
-        jsonObject.put("phone",vipOrder.getPhone());
-        jsonObject.put("userId",vipOrder.getUserId());
-        return JSON.toJSONString(jsonObject);
+    private AddUserVipRecordMessage buildVipRecordMsg(VipOrder vipOrder) {
+        AddUserVipRecordMessage message = new AddUserVipRecordMessage();
+        message.setKey(IdUtils.randomUUID());
+        message.setSource(ServiceNameConstants.ORDER_SERVICE);
+        message.setOrderNo(vipOrder.getOrderNo());
+        message.setStartTime(vipOrder.getStartTime());
+        message.setEndTime(vipOrder.getEndTime());
+        message.setVipId(vipOrder.getVipId());
+        message.setVipName(vipOrder.getVipName());
+        message.setVipTypeName(vipOrder.getVipTypeName());
+        message.setUserClientType(vipOrder.getUserClientType());
+        message.setPhone(vipOrder.getPhone());
+        message.setUserId(vipOrder.getUserId());
+        return message;
     }
 
+    @Override
+    protected void handleMessage(VipOrderPayNotifyMessage message) throws Exception {
+
+    }
+
+    @Override
+    protected void handleMaxRetriesExceeded(VipOrderPayNotifyMessage message) {
+
+    }
+
+    @Override
+    protected boolean filter(VipOrderPayNotifyMessage message) {
+        return true;
+    }
+
+    @Override
+    protected boolean isRetry() {
+        return true;
+    }
+
+    @Override
+    protected boolean throwException() {
+        return false;
+    }
 }

bnyer-services/bnyer-order/src/main/java/com/bnyer/order/service/impl/VipOrderServiceImpl.java

@@ -5,6 +5,7 @@ import cn.hutool.core.date.DateUtil;
 import com.alibaba.fastjson.JSON;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.bnyer.common.core.constant.ServiceNameConstants;
 import com.bnyer.common.core.domain.R;
 import com.bnyer.common.core.domain.VipOrder;
 import com.bnyer.common.core.enums.*;
@@ -13,8 +14,12 @@ import com.bnyer.common.core.utils.DateUtils;
 import com.bnyer.common.core.utils.OrderUtil;
 import com.bnyer.common.core.utils.StringUtils;
 import com.bnyer.common.core.utils.bean.EntityConvertUtil;
+import com.bnyer.common.core.utils.uuid.IdUtils;
 import com.bnyer.common.core.vo.UserInfoVo;
-import com.bnyer.common.rocketmq.config.RocketMqConstant;
+import com.bnyer.common.rocketmq.constant.RocketMqConstant;
+import com.bnyer.common.rocketmq.constant.RocketMqTopic;
+import com.bnyer.common.rocketmq.domain.order.VipOrderCancelMessage;
+import com.bnyer.common.rocketmq.template.RocketMQEnhanceTemplate;
 import com.bnyer.common.security.utils.SecurityUtils;
 import com.bnyer.img.api.remote.RemoteUserVipService;
 import com.bnyer.img.api.vo.UserVipVo;
@@ -27,8 +32,6 @@ import com.bnyer.common.core.enums.EnumVipOrderStatus;
 import com.bnyer.order.mapper.VipOrderMapper;
 import com.bnyer.order.service.VipOrderService;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.rocketmq.client.producer.SendCallback;
-import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.SendStatus;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -50,8 +53,12 @@ import java.util.stream.Collectors;
 @Service
 public class VipOrderServiceImpl extends ServiceImpl<VipOrderMapper, VipOrder> implements VipOrderService {
 
+//    @Autowired
+//    private RocketMQTemplate orderCancelMqTemplate;
+
+    //注入增强后的模板，可以自动实现环境隔离，日志记录
     @Autowired
-    private RocketMQTemplate orderCancelMqTemplate;
+    private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;
 
     @Autowired
     private VipOrderMapper vipOrderMapper;
@@ -107,22 +114,17 @@ public class VipOrderServiceImpl extends ServiceImpl<VipOrderMapper, VipOrder> i
         vipOrderMapper.insert(vipOrder);
         String orderNo = vipOrder.getOrderNo();
         //发送消息，如果三十分钟后没有支付，则取消订单
-        orderCancelMqTemplate.asyncSend(RocketMqConstant.VIP_ORDER_CANCEL_TOPIC, orderNo,new SendCallback() {
-            @Override
-            public void onSuccess(SendResult sendResult) {
-                log.info("topic:{}消息发送成功,result:{}",RocketMqConstant.VIP_ORDER_CANCEL_TOPIC,JSON.toJSONString(sendResult));
-            }
-            @Override
-            public void onException(Throwable e) {
-                log.info("topic:{}消息发送失败,error:{}",RocketMqConstant.VIP_ORDER_CANCEL_TOPIC,e.getMessage());
-            }
-        },RocketMqConstant.TIMEOUT);
+        VipOrderCancelMessage vipOrderCancelMessage = new VipOrderCancelMessage();
+        vipOrderCancelMessage.setKey(IdUtils.randomUUID());
+        vipOrderCancelMessage.setSource(ServiceNameConstants.ORDER_SERVICE);
+        vipOrderCancelMessage.setOrderNo(orderNo);
+        rocketMQEnhanceTemplate.sendAsyncMsg(RocketMqTopic.VIP_ORDER_CANCEL_TOPIC,null,vipOrderCancelMessage,RocketMqConstant.THIRTY_MINUTES);
 //        SendStatus sendStatus = orderCancelMqTemplate.syncSend(RocketMqConstant.VIP_ORDER_CANCEL_TOPIC, new GenericMessage<>(orderNo), RocketMqConstant.TIMEOUT, RocketMqConstant.CANCEL_ORDER_DELAY_LEVEL).getSendStatus();
 //        if (!Objects.equals(sendStatus,SendStatus.SEND_OK)) {
 //            // 消息发不出去就抛异常，发的出去无所谓
 //            throw new ServiceException(ResponseEnum.SERVER_ERROR);
 //        }else {
-//            log.info("消息发送成功,topic:{}",RocketMqConstant.VIP_ORDER_CANCEL_TOPIC);
+//            log.info("消息发送成功,topic:{}",RocketMqTopic.VIP_ORDER_CANCEL_TOPIC);
 //        }
         return orderNo;
     }

bnyer-services/bnyer-pay/src/main/java/com/bnyer/pay/config/RocketMqConfig.java

@@ -1,28 +1,28 @@
-package com.bnyer.pay.config;
-
-import com.bnyer.common.rocketmq.config.RocketMqAdapter;
-import com.bnyer.common.rocketmq.config.RocketMqConstant;
-import org.apache.rocketmq.spring.core.RocketMQTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.context.config.annotation.RefreshScope;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Lazy;
-
-/**
- * @author :WXC
- * @Date :2023/03/24
- * @description :
- */
-@RefreshScope
-@Configuration
-public class RocketMqConfig {
-    @Autowired
-    private RocketMqAdapter rocketMqAdapter;
-
-    @Lazy
-    @Bean(destroyMethod = "destroy")
-    public RocketMQTemplate vipOrderPayNotifyMqTemplate() {
-        return rocketMqAdapter.getTemplateByTopicName(RocketMqConstant.VIP_ORDER_PAY_NOTIFY_TOPIC);
-    }
-}
+//package com.bnyer.pay.config;
+//
+//import com.bnyer.common.rocketmq.adapter.RocketMqAdapter;
+//import com.bnyer.common.rocketmq.constant.RocketMqTopic;
+//import com.bnyer.common.rocketmq.template.RocketMQEnhanceTemplate;
+//import org.springframework.beans.factory.annotation.Autowired;
+//import org.springframework.cloud.context.config.annotation.RefreshScope;
+//import org.springframework.context.annotation.Bean;
+//import org.springframework.context.annotation.Configuration;
+//import org.springframework.context.annotation.Lazy;
+//
+///**
+// * @author :WXC
+// * @Date :2023/03/24
+// * @description :
+// */
+//@RefreshScope
+//@Configuration
+//public class RocketMqConfig {
+//    @Autowired
+//    private RocketMqAdapter rocketMqAdapter;
+//
+//    @Lazy
+//    @Bean(destroyMethod = "destroy")
+//    public RocketMQEnhanceTemplate vipOrderPayNotifyMqTemplate() {
+//        return rocketMqAdapter.getTemplateByTopicName(RocketMqTopic.VIP_ORDER_PAY_NOTIFY_TOPIC);
+//    }
+//}

bnyer-services/bnyer-pay/src/main/java/com/bnyer/pay/service/impl/PayInfoServiceImpl.java

@@ -1,15 +1,18 @@
 package com.bnyer.pay.service.impl;
 
 import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.bnyer.common.core.constant.ServiceNameConstants;
 import com.bnyer.common.core.domain.PayInfo;
 import com.bnyer.common.core.enums.EnumSceneCode;
 import com.bnyer.common.core.enums.ResponseEnum;
 import com.bnyer.common.core.exception.ServiceException;
 import com.bnyer.common.core.utils.bean.EntityConvertUtil;
-import com.bnyer.common.rocketmq.config.RocketMqConstant;
+import com.bnyer.common.core.utils.uuid.IdUtils;
+import com.bnyer.common.rocketmq.constant.RocketMqTopic;
+import com.bnyer.common.rocketmq.domain.order.VipOrderPayNotifyMessage;
+import com.bnyer.common.rocketmq.template.RocketMQEnhanceTemplate;
 import com.bnyer.pay.bean.dto.AddPayInfoDto;
 import com.bnyer.pay.bean.dto.EditPayInfoNotifyDto;
 import com.bnyer.pay.bean.dto.EditPayInfoSingleDto;
@@ -17,9 +20,6 @@ import com.bnyer.pay.bean.vo.PayInfoDetailsVo;
 import com.bnyer.pay.mapper.PayInfoMapper;
 import com.bnyer.pay.service.PayInfoService;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.rocketmq.client.producer.SendCallback;
-import org.apache.rocketmq.client.producer.SendResult;
-import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -38,7 +38,7 @@ public class PayInfoServiceImpl extends ServiceImpl<PayInfoMapper, PayInfo> impl
     private PayInfoMapper payInfoMapper;
 
     @Autowired
-    private RocketMQTemplate vipOrderPayNotifyMqTemplate;
+    private RocketMQEnhanceTemplate rocketMQEnhanceTemplate;
 
     /**
      * 添加支付订单
@@ -81,40 +81,16 @@ public class PayInfoServiceImpl extends ServiceImpl<PayInfoMapper, PayInfo> impl
             //会员充值场景
             case VIP_RECHARGE:
                 // 发送消息，订单支付成功
-                JSONObject vipRechargeMsgObj = new JSONObject();
-                vipRechargeMsgObj.put("orderNo",orderNo);
-                String vipRechargeMsgStr = JSON.toJSONString(vipRechargeMsgObj);
                 // TODO: 2023/04/23可优化为：添加一张消息日志表，字段：topicName，消息内容，消息状态(发送中、成功、失败),错误信息
-                sendMsg(RocketMqConstant.VIP_ORDER_PAY_NOTIFY_TOPIC,vipRechargeMsgStr);
-//                SendStatus sendStatus = vipOrderPayNotifyMqTemplate.syncSend(RocketMqConstant.VIP_ORDER_PAY_NOTIFY_TOPIC,
-//                        new GenericMessage<>(vipRechargeMsgStr)).getSendStatus();
-//                if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
-//                    // 消息发不出去就抛异常，因为订单回调会有多次，几乎不可能每次都无法发送出去，发的出去无所谓
-//                    throw new ServiceException(ResponseEnum.SERVER_ERROR);
-//                }
-//                break;
+                VipOrderPayNotifyMessage vipOrderPayNotifyMessage = new VipOrderPayNotifyMessage();
+                vipOrderPayNotifyMessage.setKey(IdUtils.randomUUID());
+                vipOrderPayNotifyMessage.setSource(ServiceNameConstants.PAY_SERVICE);
+                vipOrderPayNotifyMessage.setOrderNo(orderNo);
+                rocketMQEnhanceTemplate.sendAsyncRetryMsg(RocketMqTopic.VIP_ORDER_PAY_NOTIFY_TOPIC,null, vipOrderPayNotifyMessage);
+                break;
         }
     }
 
-    /**
-     * 发送消息
-     * @param topic
-     * @param msg
-     * @param <T>
-     */
-    public <T> void sendMsg(String topic, T msg) {
-        vipOrderPayNotifyMqTemplate.asyncSend(topic, msg,new SendCallback() {
-            @Override
-            public void onSuccess(SendResult sendResult) {
-                log.info("topic:{}消息发送成功,result:{}",topic,JSON.toJSONString(sendResult));
-            }
-            @Override
-            public void onException(Throwable e) {
-                log.info("topic:{}消息发送失败,error:{}",topic,e.getMessage());
-            }
-        });
-    }
-
     /**
      * 构建修改支付单信息实体
      * @param editPayInfoNotifyDto