init

src/main/java/com/senmesh/mq/config/ExchangeManager.java

@@ -0,0 +1,54 @@
+package com.senmesh.mq.config;
+
+import org.springframework.amqp.core.Exchange;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import com.senmesh.common.exception.util.Assert;
+import com.senmesh.mq.constant.ExchangeTypeEnum;
+import com.senmesh.mq.exception.MqException;
+
+public class ExchangeManager {
+  
+  @Autowired
+  private TopicExchange topicExchange;
+  
+//  @Autowired
+//  private CustomExchange delayedCustomExchange;
+  
+  public TopicExchange getTopicExchange() {
+    return topicExchange;
+  }
+  
+//  public CustomExchange getDelayedCustomExchange() {
+//    return delayedCustomExchange;
+//  }
+  
+  /**
+   * 获取交换器名称
+   * 
+   * @return
+   * @author sr on 2020-07-27
+   */
+  public String getTopicExchangeName() {
+    return topicExchange.getName();
+  }
+  
+  public Exchange getExchangeByType(String exchangeType) {
+    Assert.notNull(exchangeType, "交换机不能为空");
+    ExchangeTypeEnum exchangeTypeEnum = ExchangeTypeEnum.valueOf(exchangeType);
+    switch (exchangeTypeEnum) {
+      case TOPIC:
+        return topicExchange;
+//      case DELAYED:
+//        return delayedCustomExchange;
+      default:
+        throw new MqException("未找到对应交换机");
+    }
+  }
+  
+  public String getExchangeNameByType(String exchangeType) {
+    return getExchangeByType(exchangeType).getName();
+  }
+
+}

src/main/java/com/senmesh/mq/config/QueueManager.java

@@ -0,0 +1,65 @@
+package com.senmesh.mq.config;
+
+import org.springframework.amqp.core.*;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.beans.factory.annotation.Autowired;
+
+/**
+ * 消息管理（queue/binding）
+ *
+ * @author sr on 2020-07-22
+ */
+public class QueueManager {
+
+  @Autowired
+  private AmqpAdmin amqpAdmin;
+
+  @Autowired
+  private ExchangeManager exchangeManager;
+
+  @Autowired
+  private AbstractMessageListenerContainer messageListenerContainer;
+
+  /**
+   * 添加队列
+   * 
+   * @param queueName
+   * @return
+   * @author sr on 2020-07-27
+   */
+  public Queue addQueue(String queueName) {
+    Queue queue = new Queue(queueName);
+    if (amqpAdmin.getQueueProperties(queueName) == null) {
+      amqpAdmin.declareQueue(queue);
+    }
+    return queue;
+  }
+
+  /**
+   * 绑定队列和路由键
+   *
+   * @param queueName
+   * @param routingKey
+   * @author sr on 2020-07-27
+   */
+  public void addQueueAndRoutingKey(String queueName, String routingKey) {
+    Queue queue = this.addQueue(queueName);
+    Binding binding = BindingBuilder.bind(queue).to(exchangeManager.getTopicExchange()).with(routingKey);
+    amqpAdmin.declareBinding(binding);
+    messageListenerContainer.addQueues(queue);
+  }
+
+  /**
+   * 绑定队列和路由键
+   *
+   * @param queueName
+   * @param routingKey
+   * @author sr on 2020-07-27
+   */
+  public void addQueueAndRoutingKey(Exchange exchange, String queueName, String routingKey) {
+    Queue queue = this.addQueue(queueName);
+    amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(routingKey).noargs());
+    messageListenerContainer.addQueues(queue);
+  }
+
+}

src/main/java/com/senmesh/mq/config/RabbitCustomProperties.java

@@ -0,0 +1,37 @@
+package com.senmesh.mq.config;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * 定制配置
+ *
+ * @author sr on 2020-07-22
+ */
+@ConfigurationProperties(prefix = "rabbitmq.custom")
+@Data
+@NoArgsConstructor
+public class RabbitCustomProperties {
+
+  /**
+   * 通配符交换器名称
+   **/
+  private String topicExchangeName;
+
+  /**
+   * 延时队里名称
+   */
+  private String delayedExchangeName;
+
+  /**
+   * 指定消费者ip
+   **/
+  private String consumerAppId;
+
+  /**
+   * 广播id-同系统内需要广播时配置
+   **/
+  private String broadcastId;
+
+}

src/main/java/com/senmesh/mq/config/RabbitMqConfiguration.java

@@ -0,0 +1,194 @@
+package com.senmesh.mq.config;
+
+import java.util.List;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import com.senmesh.common.CommonConfiguration;
+import com.senmesh.common.constant.CommonConstants;
+import com.senmesh.common.util.EnvironmentUtils;
+import com.senmesh.mq.consumer.ConsumerDeliverer;
+import com.senmesh.mq.consumer.InitHandler;
+import com.senmesh.mq.consumer.handler.IMqHandler;
+import com.senmesh.mq.producer.sender.DefaultMqSender;
+import com.senmesh.mq.producer.sender.IMqSender;
+import com.senmesh.mq.util.MqSendUtils;
+
+/**
+ * @author sr on 2020-07-22
+ */
+@Configuration
+@EnableConfigurationProperties(RabbitCustomProperties.class)
+@AutoConfigureAfter(CommonConfiguration.class)
+public class RabbitMqConfiguration {
+
+  public static final String DEFAULT_EXCHANGE_NAME = "alita_exchange";
+
+  /**
+   * 默认延时队列交换机名称
+   */
+  public static final String DEFAULT_DELAYED_EXCHANGE_NAME = "alita_delayed_exchange";
+
+  @Autowired
+  private RabbitCustomProperties rabbitCustomProperties;
+
+  private String getExchangeName() {
+    String exchangeName = rabbitCustomProperties.getTopicExchangeName();
+    if (StringUtils.isBlank(exchangeName)) {
+      exchangeName = DEFAULT_EXCHANGE_NAME;
+    }
+    return String.join(CommonConstants.UNDERLINE, exchangeName,
+        EnvironmentUtils.getProfile());
+  }
+
+  private String getDelayedExchangeName() {
+    String delayedExchangeName = rabbitCustomProperties.getDelayedExchangeName();
+    if (StringUtils.isEmpty(delayedExchangeName)) {
+      delayedExchangeName = DEFAULT_DELAYED_EXCHANGE_NAME;
+    }
+    return String.join(CommonConstants.UNDERLINE, delayedExchangeName,
+        EnvironmentUtils.getProfile());
+  }
+
+
+  /**
+   * 交换器
+   *
+   * @return
+   * @author sr on 2020-07-22
+   */
+  @Bean
+  TopicExchange topicExchange(EnvironmentUtils environmentUtils) {
+    return new TopicExchange(getExchangeName());
+  }
+
+  /*@Bean
+  public CustomExchange delayedCustomExchange(EnvironmentUtils environmentUtils) {
+    Map<String, Object> args = Maps.newHashMap();
+    //延时交换机一定要设置x-delayed-type属性
+    args.put("x-delayed-type", "direct");
+    //第二个参数type需要设置成x-delayed-message
+    return new CustomExchange(getDelayedExchangeName(), "x-delayed-message", true, false, args);
+  }*/
+
+  /**
+   * 消息监听容器
+   *
+   * @param containerFactory
+   * @param messageListenerAdapter
+   * @return
+   * @author sr on 2020-07-22
+   */
+  @Bean
+  <T extends AbstractMessageListenerContainer> T messageListenerContainer(
+      AbstractRabbitListenerContainerFactory<T> containerFactory,
+      MessageListenerAdapter messageListenerAdapter) {
+    AbstractMessageListenerContainer container =
+        containerFactory.createListenerContainer();
+    container.setMessageListener(messageListenerAdapter);
+    return (T) container;
+  }
+
+  /**
+   * 消息监听适配器
+   *
+   * @param consumerDeliverer
+   * @return
+   * @author sr on 2020-07-22
+   */
+  @Bean
+  MessageListenerAdapter messageListenerAdapter(ConsumerDeliverer consumerDeliverer) {
+    MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(consumerDeliverer);
+    // 暂不做消息转换, 交给consumerDeliverer处理
+    messageListenerAdapter.setMessageConverter(null);
+    return messageListenerAdapter;
+  }
+
+  /**
+   * 初始化消息处理器，注册队列，绑定路由
+   * 
+   * @param messageManager
+   * @param handlers
+   * @return
+   * @author sr on 2020-07-24
+   */
+  @Bean
+  InitHandler initHandler() {
+    return new InitHandler();
+  }
+
+  /**
+   * 消息转换器
+   * 
+   * @return
+   * @author sr on 2020-07-24
+   */
+  @Bean
+  Jackson2JsonMessageConverter jackson2JsonMessageConverter() {
+    return new Jackson2JsonMessageConverter("com.newyetai");
+  }
+
+  /**
+   * 队列管理
+   * 
+   * @return
+   * @author sr on 2020-07-24
+   */
+  @Bean
+  QueueManager queueManager() {
+    return new QueueManager();
+  }
+  
+  /**
+   * 交换机管理
+   * @return
+   * @author sr on 2021-01-06 
+   */
+  @Bean
+  ExchangeManager exchangeManager() {
+    return new ExchangeManager();
+  }
+
+  /**
+   * 消息消费分发
+   *
+   * @param handlers
+   * @param handlers
+   * @return
+   * @author sr on 2020-07-24
+   */
+  @Bean
+  ConsumerDeliverer consumerDeliverer(List<IMqHandler<?>> handlers) {
+    ConsumerDeliverer consumerDeliverer = new ConsumerDeliverer(handlers);
+    consumerDeliverer.setMessageConvert(jackson2JsonMessageConverter());
+    return consumerDeliverer;
+  }
+
+  /**
+   * 消息发送者
+   * 
+   * @param rabbitTemplate
+   * @param messageManager
+   * @return
+   * @author sr on 2020-07-24
+   */
+  @Bean
+  IMqSender mqSender(RabbitTemplate rabbitTemplate, QueueManager messageManager,
+      RabbitCustomProperties rabbitCustomProperties) {
+    DefaultMqSender defaultMqSender =  new DefaultMqSender();
+    MqSendUtils mqSendUtils = new MqSendUtils(defaultMqSender);
+    return defaultMqSender;
+  }
+
+}

src/main/java/com/senmesh/mq/constant/ExchangeTypeEnum.java

@@ -0,0 +1,15 @@
+package com.senmesh.mq.constant;
+
+/**
+ * 交换机
+ * @author sr on 2021-01-06 
+ */
+public enum ExchangeTypeEnum {
+  
+  /** (通用)主题交换机 **/
+  TOPIC,
+  /** 延迟交换机 **/
+  DELAYED;
+  
+
+}

src/main/java/com/senmesh/mq/constant/MqConstant.java

@@ -0,0 +1,12 @@
+package com.senmesh.mq.constant;
+
+/**
+ * mq常数
+ *
+ * @author Carlye
+ * @date 2021/01/14
+ */
+public interface MqConstant {
+
+    String NACOS_DISCOVERY_IP = "spring.cloud.nacos.discovery.ip";
+}

src/main/java/com/senmesh/mq/consumer/ConsumerDeliverer.java

@@ -0,0 +1,138 @@
+package com.senmesh.mq.consumer;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.core.ParameterizedTypeReference;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.senmesh.common.constant.ProjectProfileEnum;
+import com.senmesh.common.util.EnvironmentUtils;
+import com.senmesh.common.util.JsonUtils;
+import com.senmesh.common.util.ParameterizedTypeImpl;
+import com.senmesh.mq.constant.MqConstant;
+import com.senmesh.mq.consumer.handler.IMqHandler;
+import com.senmesh.mq.dto.IMqMessage;
+import com.rabbitmq.client.Channel;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * 消息分发
+ *
+ * @author sr on 2020-07-22
+ */
+@Slf4j
+public class ConsumerDeliverer implements ChannelAwareMessageListener {
+
+  private Jackson2JsonMessageConverter jackson2JsonMessageConverter =
+      new Jackson2JsonMessageConverter();
+
+  public static final Map<String, IMqHandler<?>> MQ_HANDLER_MAP = Maps.newHashMap();
+
+  /**
+   * key-routingKey,value-ParameterizedTypeReference
+   **/
+  @SuppressWarnings("rawtypes")
+  private Map<String, ParameterizedTypeReference> messqgeTypeMap = Maps.newConcurrentMap();
+
+  public ConsumerDeliverer(List<IMqHandler<?>> handlers) {
+    handlers.forEach(handler -> MQ_HANDLER_MAP.put(handler.getRoutingKey(), handler));
+  }
+
+  public void setMessageConvert(Jackson2JsonMessageConverter jackson2JsonMessageConverter) {
+    this.jackson2JsonMessageConverter = jackson2JsonMessageConverter;
+  }
+
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  @Override
+  public void onMessage(Message message, Channel channel) throws Exception {
+    String routingKey = message.getMessageProperties().getReceivedRoutingKey();
+    String appId = StringUtils.defaultString(message.getMessageProperties().getAppId(), "");
+    Set<String> appIp = Sets.newHashSet(appId.split(","));
+    //TODO 重复失败消费三次，进死信
+
+    IMqHandler handler = MQ_HANDLER_MAP.get(routingKey);
+    if (handler == null) {
+      log.info(routingKey + "未找到对应routingKey消费者...");
+      return;
+    }
+    
+    // 非广播消息&指定消费者ip消费&只能开发环境
+    if (!handler.isBroadcast()
+        && ProjectProfileEnum.DEV.getValue().equals(EnvironmentUtils.getProfile())
+        && StringUtils.isNotBlank(appId)) {
+      // String ip = InetAddress.getLocalHost().getHostAddress();
+      String ip = EnvironmentUtils.getProperty(MqConstant.NACOS_DISCOVERY_IP);
+      log.info("Mq消息消费指定appId：{}，当前ip：{}", appId, ip);
+      if (!appIp.contains(ip)) {
+        log.info("Mq消息消费异常,appId:[{}],ip:[{}],message:[{}]", appId, ip, JsonUtils.toJsonString(message));
+        //放回queue使用其他消费者消费
+        channel.basicRecover(true);
+        return;
+      }
+    }
+    
+    Object obj = null;
+    try {
+      obj = jackson2JsonMessageConverter.fromMessage(message,
+          getParameterizedTypeReference(routingKey, handler));
+      log.info("mqDeliverer msg:{}", message);
+      handler.onMessage((IMqMessage) obj);
+      return;
+    } catch (MessageConversionException e) {
+      // 转换失败异常处理
+      log.info("mqMessage convert error:{}", e);
+    } catch (Exception e) {
+      // TODO 消息消费异常处理
+      log.info("mqMessage deal obj:{}", obj, e);
+      //throw new MqException("message consume error:" + e.getMessage());
+    }
+  }
+  
+  /**
+   * 获取消息范型对象
+   * 
+   * @param routingKey
+   * @param handler
+   * @return
+   * @author sr on 2020-09-02
+   */
+  @SuppressWarnings("rawtypes")
+  private ParameterizedTypeReference getParameterizedTypeReference(String routingKey,
+      IMqHandler handler) {
+    ParameterizedTypeReference type = messqgeTypeMap.get(routingKey);
+    if (type != null) {
+      return type;
+    }
+    // 有可能是代理对象
+    for(Method method :AopUtils.getTargetClass(handler).getMethods()) {
+      // 有可能是同名方法,非桥接
+      if ("onMessage".equals(method.getName()) && !method.isBridge()) {
+        try {
+          Type argType = method.getGenericParameterTypes()[0];
+          ParameterizedType parameterizedType = (ParameterizedType) (argType);
+          type = ParameterizedTypeReference.forType(new ParameterizedTypeImpl(
+              new Type[] {parameterizedType.getActualTypeArguments()[0]}, null, parameterizedType.getRawType()));
+          messqgeTypeMap.put(routingKey, type);
+          break;
+        } catch (Exception e) {
+          log.error(e.getMessage());
+        }
+      }
+    }
+    return type;
+  }
+
+}

src/main/java/com/senmesh/mq/consumer/InitHandler.java

@@ -0,0 +1,56 @@
+package com.senmesh.mq.consumer;
+
+import com.google.common.collect.Lists;
+import com.senmesh.common.constant.CommonConstants;
+import com.senmesh.common.util.EnvironmentUtils;
+import com.senmesh.mq.config.ExchangeManager;
+import com.senmesh.mq.config.QueueManager;
+import com.senmesh.mq.config.RabbitCustomProperties;
+import com.senmesh.mq.constant.MqConstant;
+import com.senmesh.mq.consumer.handler.IMqHandler;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.List;
+
+/**
+ * 初始化消息消费处理类，初始queue/binding
+ * 
+ * @author sr on 2020-07-24
+ */
+public class InitHandler implements InitializingBean {
+
+  @Autowired(required = false)
+  private List<IMqHandler<?>> handlers = Lists.newArrayList();
+
+  @Autowired
+  private QueueManager queueManager;
+  
+  @Autowired 
+  private ExchangeManager exchangeManager;
+  
+  @Autowired
+  private RabbitCustomProperties rabbitCustomProperties;
+
+  @Override
+  public void afterPropertiesSet() throws Exception {
+    String applicationName = EnvironmentUtils.getApplicationName();
+    handlers.forEach(handler -> {
+      String routingKey = handler.getRoutingKey();
+      String queueName = handler.getQueueName();
+      // 初始化binding
+      if (queueName == null) {
+        queueName = String.join(CommonConstants.UNDERLINE, applicationName,
+            EnvironmentUtils.getProfile(), routingKey);
+        if (handler.isBroadcast()) {
+//          queueName = String.join(CommonConstants.UNDERLINE, queueName,rabbitCustomProperties.getBroadcastId());
+          String nacosDiscoveryIp = EnvironmentUtils.getProperty(MqConstant.NACOS_DISCOVERY_IP);
+          queueName = String.join(CommonConstants.UNDERLINE, queueName, nacosDiscoveryIp);
+        }
+      }
+      queueManager.addQueueAndRoutingKey(
+          exchangeManager.getExchangeByType(handler.getExchangeType()), queueName, routingKey);
+    });
+  }
+  
+}

src/main/java/com/senmesh/mq/consumer/handler/IDelayedMqHandler.java

@@ -0,0 +1,22 @@
+//package com.newyetai.mq.consumer.handler;
+//
+//import com.newyetai.mq.constant.ExchangeTypeEnum;
+//import com.newyetai.mq.dto.DelayedMqMessage;
+//
+///**
+// * 消费者处理类(延时交换机)
+// * 
+// * @author sr on 2021-01-06 
+// * @param <T>
+// */
+//public interface IDelayedMqHandler<T> extends IMqHandler<DelayedMqMessage<T>>{
+//  
+//  default String getExchangeType() {
+//    return ExchangeTypeEnum.DELAYED.name();
+//  }
+//  
+//  default Class<DelayedMqMessage> getMessageTypeClazz() {
+//    return DelayedMqMessage.class;
+//  }
+//
+//}

src/main/java/com/senmesh/mq/consumer/handler/IMqHandler.java

@@ -0,0 +1,67 @@
+package com.senmesh.mq.consumer.handler;
+
+import com.senmesh.mq.constant.ExchangeTypeEnum;
+import com.senmesh.mq.dto.IMqMessage;
+
+/**
+ * 消费者处理类
+ * 
+ * @author sr on 2021-01-06 
+ * @param <T>
+ */
+public interface IMqHandler<M extends IMqMessage<?>> {
+  
+  /**
+   * 为空，默认:系统名_routingKey；每个routingkey有专属queue
+   * 
+   * @return
+   * @author sr on 2020-07-24
+   */
+  default String getQueueName() {
+    return null;
+  }
+  
+  /**
+   * 是否同项目广播，默认否
+   * @return
+   * @author sr on 2020-08-04 
+   */
+  default boolean isBroadcast() {
+    return false;
+  }
+
+  /**
+   * routingKey 路由键
+   * 
+   * @return
+   * @author sr on 2020-07-24
+   */
+  String getRoutingKey();
+
+  /**
+   * 消费消息
+   *
+   * @param message
+   * @return 如果抛异常，则代表消费失败，会重新入队
+   * @author sr on 2020-07-24
+   */
+  void onMessage(M message);
+  
+  /**
+   * 获取mq消息对象class
+   * 
+   * @return
+   * @author sr on 2021-01-06 
+   */
+  Class<? extends IMqMessage> getMessageTypeClazz();
+  
+
+  /**
+   * 交换机
+   * 
+   * @return {@link ExchangeTypeEnum}.name()
+   * @author sr on 2021-01-06 
+   */
+  String getExchangeType();
+
+}

src/main/java/com/senmesh/mq/consumer/handler/ITopicMqHandler.java

@@ -0,0 +1,28 @@
+package com.senmesh.mq.consumer.handler;
+
+import com.senmesh.mq.constant.ExchangeTypeEnum;
+import com.senmesh.mq.dto.TopicMqMessage;
+
+/**
+ * 消费者处理类(通用交换机)
+ * 
+ * @author sr on 2020-07-22
+ * @param <T>
+ */
+public interface ITopicMqHandler<T> extends IMqHandler<TopicMqMessage<T>>{
+
+  default Class<TopicMqMessage> getMessageTypeClazz() {
+    return TopicMqMessage.class;
+  }
+
+  /**
+   * 交换机
+   * 
+   * @return {@link ExchangeTypeEnum}.name()
+   * @author sr on 2021-01-06 
+   */
+  default String getExchangeType() {
+    return ExchangeTypeEnum.TOPIC.name();
+  }
+  
+}

src/main/java/com/senmesh/mq/dto/AbstractMqMessage.java

@@ -0,0 +1,38 @@
+package com.senmesh.mq.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@SuperBuilder
+public abstract class AbstractMqMessage<T> implements IMqMessage<T>{
+  
+  /**
+   * 消息id，可为空（uuid生成）
+   **/
+  private String messageId;
+
+  /**
+   * 路由键，不能为空
+   **/
+  private String routingKey;
+  
+  /**
+   * 消息对象
+   **/
+  private T data;
+  
+  /** 操作机构ID(发送方有登录信息时有值) **/
+  private Long operateOrgId;
+  
+  /** 操作用户ID(发送方有登录信息时有值) **/
+  private Long operateUserId;
+
+  /** 用户类型 **/
+  private Integer operateUserType;
+
+}

src/main/java/com/senmesh/mq/dto/DelayedMqMessage.java

@@ -0,0 +1,39 @@
+package com.senmesh.mq.dto;
+
+import com.senmesh.mq.constant.ExchangeTypeEnum;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder.Default;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+@Data
+@ToString(callSuper = true)
+@NoArgsConstructor
+@AllArgsConstructor
+@SuperBuilder
+public class DelayedMqMessage<T> extends AbstractMqMessage<T>{
+  
+  /** 交换机,可为空(DELAYED) **/
+  @Default
+  private String exchangeType = ExchangeTypeEnum.DELAYED.name();
+
+//  /**
+//   * 延时发送时间,不能为空
+//   */
+//  private Long delayedTime;
+//
+//  /**
+//   * 时间类型,可为空
+//   */
+//  @Default
+//  private TimeUnit timeUnit = TimeUnit.SECONDS;
+//
+//  
+//  public TimeUnit getTimeUnit() {
+//    return Optional.ofNullable(timeUnit).orElse(TimeUnit.SECONDS);
+//  }
+
+}

src/main/java/com/senmesh/mq/dto/IMqMessage.java

@@ -0,0 +1,35 @@
+package com.senmesh.mq.dto;
+
+import com.senmesh.mq.constant.ExchangeTypeEnum;
+
+/**
+ * 消息
+ * 
+ * @author sr on 2021-01-06 
+ * @param <T>
+ */
+public interface IMqMessage<T> {
+  
+  /**
+   * 交换机类型
+   * @return {@link ExchangeTypeEnum}
+   * @author sr on 2021-01-06 
+   */
+  String getExchangeType();
+  
+  /**
+   * routingKey 路由键
+   * 
+   * @return
+   * @author sr on 2021-01-06 
+   */
+  String getRoutingKey();
+  
+  /**
+   * 消息体
+   * @return
+   * @author sr on 2021-01-06 
+   */
+  T getData();
+  
+}

src/main/java/com/senmesh/mq/dto/TopicMqMessage.java

@@ -0,0 +1,26 @@
+package com.senmesh.mq.dto;
+
+import com.senmesh.mq.constant.ExchangeTypeEnum;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder.Default;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+/**
+ * @author sr on 2020-07-24
+ */
+@Data
+@ToString(callSuper = true)
+@NoArgsConstructor
+@AllArgsConstructor
+@SuperBuilder
+public class TopicMqMessage<T> extends AbstractMqMessage<T>{
+
+  /** 交换机 **/
+  @Default
+  private String exchangeType = ExchangeTypeEnum.TOPIC.name();
+
+}

src/main/java/com/senmesh/mq/exception/MqException.java

@@ -0,0 +1,29 @@
+package com.senmesh.mq.exception;
+
+import com.senmesh.common.exception.BusinessException;
+
+/**
+ * 异常
+ * 
+ * @author sr on 2020-07-24
+ */
+public class MqException extends BusinessException {
+
+  /**  **/
+  private static final long serialVersionUID = -7022341715833999512L;
+
+  public MqException() {}
+
+  public MqException(String message) {
+    super(message);
+  }
+
+  public MqException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public MqException(Throwable cause) {
+    super(cause);
+  }
+
+}

src/main/java/com/senmesh/mq/producer/sender/DefaultMqSender.java

@@ -0,0 +1,92 @@
+package com.senmesh.mq.producer.sender;
+
+import com.senmesh.common.AlitaResult;
+import com.senmesh.common.exception.BusinessException;
+import com.senmesh.common.exception.util.Assert;
+import com.senmesh.common.util.UserContextHolder;
+import com.senmesh.mq.config.ExchangeManager;
+import com.senmesh.mq.config.RabbitCustomProperties;
+import com.senmesh.mq.dto.AbstractMqMessage;
+import com.senmesh.mq.dto.TopicMqMessage;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author sr on 2020-07-24
+ */
+@Slf4j
+public class DefaultMqSender implements IMqSender {
+
+  @Autowired
+  private RabbitTemplate rabbitTemplate;
+
+  @Autowired
+  private RabbitCustomProperties rabbitCustomProperties;
+  
+  @Autowired
+  private ExchangeManager exchangeManager;
+  
+  @Override
+  public <T> AlitaResult<?> send(String routingKey, T msg){
+    return this.send(TopicMqMessage.builder().routingKey(routingKey).data(msg).build());
+  }
+  
+  
+  @Override
+  public <T> AlitaResult<?> sendDelayed(String routingKey, T msg, Long delayedTime, TimeUnit timeUnit) {
+    throw new BusinessException("延迟队列不可用");
+//    return this.send(DelayedMqMessage.builder().routingKey(routingKey).data(msg)
+//        .delayedTime(delayedTime).timeUnit(timeUnit).build());
+  }
+
+  @Override
+  public <T> AlitaResult<?> send(AbstractMqMessage<T> iMqMessage) {
+    try {
+      iMqMessage.setOperateUserId(UserContextHolder.getLoginUserId());
+      iMqMessage.setOperateOrgId(UserContextHolder.getLoginOrgId());
+      iMqMessage.setOperateUserType(UserContextHolder.getUserType());
+      String routingKey = iMqMessage.getRoutingKey();
+      Assert.notBlank(routingKey, "routingKey不能为空");
+      // 获取 exchange
+      String exchangeName = exchangeManager.getExchangeNameByType(iMqMessage.getExchangeType());
+      
+      MessageProperties messageProperties = new MessageProperties();
+      // 指定消费者
+      String consumerAppId = rabbitCustomProperties.getConsumerAppId();
+      if (StringUtils.isNotBlank(consumerAppId)) {
+        messageProperties.setAppId(consumerAppId);
+      }
+      // 处理延时
+//      this.dealDelayed(iMqMessage, messageProperties);
+      
+      Message msg = rabbitTemplate.getMessageConverter().toMessage(iMqMessage, messageProperties);
+      log.info("mqSender msg:{}", msg);
+      rabbitTemplate.send(exchangeName, routingKey, msg);
+    } catch (AmqpException e) {
+      log.info("mqSender send error:{}", e.getMessage());
+      return AlitaResult.bizError(e.getMessage());
+    }
+    return AlitaResult.success();
+  }
+  
+  /*private void dealDelayed(IMqMessage<?> mqMessage, MessageProperties messageProperties) {
+    // 延时消息设置
+    if (mqMessage instanceof DelayedMqMessage) {
+      Long delayedTime = ((DelayedMqMessage) mqMessage).getDelayedTime();
+      Assert.notNull(delayedTime, "延时时间不能为空");
+      TimeUnit timeUnit = ((DelayedMqMessage) mqMessage).getTimeUnit();
+      if (delayedTime != null && delayedTime > 0L) {
+        messageProperties.setHeader("x-delay", timeUnit.toMillis(delayedTime));
+      }
+    }
+  }*/
+
+
+}

src/main/java/com/senmesh/mq/producer/sender/IMqSender.java

@@ -0,0 +1,50 @@
+package com.senmesh.mq.producer.sender;
+
+import java.util.concurrent.TimeUnit;
+
+import com.senmesh.common.AlitaResult;
+import com.senmesh.mq.dto.AbstractMqMessage;
+
+/**
+ * 消息发送者
+ * 
+ * @author sr on 2020-07-24
+ */
+public interface IMqSender {
+
+  /**
+   * 消息发送
+   * 
+   * @param <T>
+   * @param routingKey
+   * @param msg
+   * @return
+   * @author sr on 2020-07-24
+   */
+  <T> AlitaResult<?> send(String routingKey, T msg);
+  
+  /**
+   * 发送延迟消息
+   * 
+   * @param <T>
+   * @param routingKey
+   * @param msg
+   * @param delayedTime
+   * @param timeUnit
+   * @return
+   * @author sr on 2021-01-06 
+   */
+  <T> AlitaResult<?> sendDelayed(String routingKey, T msg, Long delayedTime, TimeUnit timeUnit);
+
+
+  /**
+   * 消息发送
+   * 
+   * @param <T>
+   * @param iMqMessage
+   * @return
+   * @author sr on 2020-07-24
+   */
+  <T> AlitaResult<?> send(AbstractMqMessage<T> iMqMessage);
+
+}

src/main/java/com/senmesh/mq/util/MqSendUtils.java

@@ -0,0 +1,62 @@
+package com.senmesh.mq.util;
+
+import com.senmesh.common.AlitaResult;
+import com.senmesh.mq.dto.AbstractMqMessage;
+import com.senmesh.mq.producer.sender.IMqSender;
+
+/**
+ * @Author dylanyuan
+ * @Date 2020/12/1 下午1:58
+ * @Version 1.0
+ */
+public class MqSendUtils {
+
+  private static IMqSender iMqSender;
+
+  public MqSendUtils(IMqSender iMqSender) {
+    MqSendUtils.iMqSender = iMqSender;
+  }
+
+  /**
+   * 消息发送
+   *
+   * @param <T>
+   * @param routingKey
+   * @param msg
+   * @return
+   * @author sr on 2020-07-24
+   */
+  public static <T> AlitaResult<?> send(String routingKey, T msg) {
+    return iMqSender.send(routingKey, msg);
+  }
+
+  /**
+   * 发送延迟消息
+   * 
+   * @param <T>
+   * @param routingKey
+   * @param msg
+   * @param delayedTime
+   * @param timeUnit
+   * @return
+   * @author sr on 2021-01-06
+   */
+//  public static <T> AlitaResult<?> sendDelayed(String routingKey, T msg, Long delayedTime,
+//      TimeUnit timeUnit) {
+//    return iMqSender.sendDelayed(routingKey, msg, delayedTime, timeUnit);
+//  }
+
+
+  /**
+   * 消息发送
+   * 
+   * @param <T>
+   * @param iMqMessage
+   * @return
+   * @author sr on 2020-07-24
+   */
+  public static <T> AlitaResult<?> send(AbstractMqMessage<T> iMqMessage) {
+    return iMqSender.send(iMqMessage);
+  }
+
+}