自定义消息队列组件 mq-spring-boot-starter

╰+哭是因爲堅強的太久メ 2021-09-24 12:12 339阅读 0赞

### 目录 ###

*   *  为什么要自定义
     *  准备工作
     *  架构构思
     *  消息对象抽取
     *  消息订阅和发布者抽取
     *  如何使用

## 为什么要自定义 ##

**为什么要自定义消息队列组件 mq-spring-boot-starter ？**

考虑到公司同一套系统能支持阿里云上部署和客户内网部署，业务代码不修改的情况下，只修改yml文件配置属性即可迁移。

本文主要针对RocketMq 进行构建，后期增加 RabbitMq的支持。

**本组件源码：**  
**[https://github.com/dwhgygzt/mq-spring-boot-starter][https_github.com_dwhgygzt_mq-spring-boot-starter]**

## 准备工作 ##

首先请确保你已经掌握了SpringBoot 如何自定义一个starter，还没有明白的请查看博文：  
[https://blog.csdn.net/gzt19881123/article/details/106456362][https_blog.csdn.net_gzt19881123_article_details_106456362]

## 架构构思 ##

想要做到业务代码不修改，那么必须创建一套自己的消息服务接口和属性对象，封装具体的消息队列产品业务逻辑。  
充分比对和分析各消息队列产品的相同业务逻辑，抽取出公共接口。

RocketMq还有半消息机制需要考虑进去。

工程整体目录结构如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d6dDE5ODgxMTIz_size_16_color_FFFFFF_t_70]

## 消息对象抽取 ##

这里目前只针对 主题类消息（Topic）进行架构，创建出自己的 消息对象。  
主要参考 ApacheRocketMq 的源码

首先创建 消息体类的公共接口：

package com.middol.starter.mq.pojo;
 
 import java.io.Serializable;
 
 /** * 公共抽象消息体 Message * * @author <a href="mailto:guzhongtao@middol.com">guzhongtao</a> */
 public interface Message extends Serializable { 
 
 /** * 获得消息体ID * @return String */
 String getMessageId();
 
 
 /** * 设置消息体ID * * @param messageId 消息唯一id */
 void setMessageId(String messageId);
 
 /** * 获得消息体 * * @return byte[] */
 byte[] getMessageBody();
 
 /** * 设置消息体 * * @param messageBody 消息体 byte[] */
 void setMessageBody(byte[] messageBody);
 
 
 }

然后创建 主题类消息体对象

package com.middol.starter.mq.pojo;
 
 import java.util.Properties;
 
 /** * 公共Topic消息体 Message * * @author <a href="mailto:guzhongtao@middol.com">guzhongtao</a> */
 public class TopicMessage implements Message { 
 private static final long serialVersionUID = 1L;
 
 /** * 用户其他属性 */
 private Properties userProperties;
 
 /** * * 消息唯一主键. * * * * 由具体mq产品生成 * */
 private String messageId;
 
 /** * * 消息主题名称, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成. * * * * 一条合法消息本成员变量不能为空 * */
 private String topicName;
 
 /** * * 消息标签, 合法标识符, 尽量简短且见名知意. * * * * 建议传递该值 * */
 private String tags;
 
 /** * * 业务主键，例如商户订单号. * * * * 建议传递该值 * */
 private String bussinessKey;
 
 /** * * 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述. * * * * 一条合法消息本成员变量不能为空 * */
 private byte[] messageBody;
 
 /** * 默认构造函数; 必要属性后续通过Set方法设置. */
 public TopicMessage() { 
 this(null, null, "", null);
 }
 
 /** * 有参构造函数. * @param topicName 消息主题 * @param tags 消息标签 * @param bussinessKey 业务主键 * @param messageBody 消息体 */
 public TopicMessage(String topicName, String tags, String bussinessKey, byte[] messageBody) { 
 this.topicName = topicName;
 this.tags = tags;
 this.bussinessKey = bussinessKey;
 this.messageBody = messageBody;
 }
 
 /** * 有参构造函数. * @param messageId 唯一主键 * @param topicName 消息主题 * @param tags 消息标签 * @param bussinessKey 业务主键 * @param messageBody 消息体 */
 public TopicMessage(String messageId,String topicName, String tags, String bussinessKey, byte[] messageBody) { 
 this.messageId = messageId;
 this.topicName = topicName;
 this.tags = tags;
 this.bussinessKey = bussinessKey;
 this.messageBody = messageBody;
 }
 
 @Override
 public String toString() { 
 return "TopicMessage [topicName=" + topicName + ", tags=" + tags + ", messageBody=" + (messageBody != null ? messageBody.length : 0) + "]";
 }
 
 public String getTopicName() { 
 return topicName;
 }
 
 public void setTopicName(String topicName) { 
 this.topicName = topicName;
 }
 
 public String getTags() { 
 return tags;
 }
 
 public void setTags(String tags) { 
 this.tags = tags;
 }
 
 public String getBussinessKey() { 
 return bussinessKey;
 }
 
 public void setBussinessKey(String bussinessKey) { 
 this.bussinessKey = bussinessKey;
 }
 
 @Override
 public String getMessageId() { 
 return messageId;
 }
 
 @Override
 public void setMessageId(String messageId) { 
 this.messageId = messageId;
 }
 
 @Override
 public byte[] getMessageBody() { 
 return messageBody;
 }
 
 @Override
 public void setMessageBody(byte[] messageBody) { 
 this.messageBody = messageBody;
 }
 
 public Properties getUserProperties() { 
 return userProperties;
 }
 
 public void setUserProperties(Properties userProperties) { 
 this.userProperties = userProperties;
 }
 
 /** * 添加用户自定义属性键值对; 该键值对在消费消费时可被获取. * @param key 自定义键 * @param value 对应值 */
 public void putUserProperties(final String key, final String value) { 
 if (null == this.userProperties) { 
 this.userProperties = new Properties();
 }
 
 if (key != null && value != null) { 
 this.userProperties.put(key, value);
 }
 }
 
 /** * 获取用户自定义键的值 * @param key 自定义键 * @return 用户自定义键值 */
 public String getUserProperties(final String key) { 
 if (null != this.userProperties) { 
 return (String) this.userProperties.get(key);
 }
 
 return null;
 }
 }

当然还包括 消息处理的状态对象：

package com.middol.starter.mq.pojo;
 
 /** * 消费消息的返回结果 * * @author <a href="mailto:guzhongtao@middol.com">guzhongtao</a> */
 public enum MessageStatus { 
 /** * 消费成功，继续消费下一条消息 */
 CommitMessage,
 /** * 消费失败，告知服务器稍后再投递这条消息，继续消费其他消息 */
 ReconsumeLater,
 }

## 消息订阅和发布者抽取 ##

订阅者

package com.middol.starter.mq.service;
 
 import com.middol.starter.mq.pojo.MessageStatus;
 import com.middol.starter.mq.pojo.TopicMessage;
 
 /** * MQ对应的监听者，实现具体的消费业务 * 订阅（subscribe）模式. * 订阅关系一致 https://help.aliyun.com/document_detail/43523.html?spm=a2c4g.11186623.6.734.60b94c07Uwhsky * 1.订阅的 Topic 必须一致 * 2.订阅的 Topic 中的 Tag 必须一致 * * @author <a href="mailto:guzhongtao@middol.com">guzhongtao</a> */
 public interface TopicListener { 
 
 /** * 对应的消费者，例如 aliyunrocketmq 中的groupId * * @return SubscriberBeanName */
 String getSubscriberBeanName();
 
 /** * 订阅的topic * * @return topic */
 String getTopicName();
 
 /** * 订阅的 tag * * @return 订阅过滤表达式字符串，ONS服务器依据此表达式进行过滤。只支持或运算 * eg: "tag1 || tag2 || tag3" * 如果subExpression等于null或者*，则表示全部订阅 */
 String getTagExpression();
 
 /** * 消息订阅 * * @param topicMessage 从消息服务器获得的订阅消息 * @return 执行完本地业务逻辑反馈消息服务器是否消费完毕 MessageStatus */
 MessageStatus subscribe(TopicMessage topicMessage);
 
 }

发布者

package com.middol.starter.mq.service;
 
 import com.middol.starter.mq.pojo.TopicMessage;
 import com.middol.starter.mq.pojo.TopicMessageSendResult;
 
 /** * 推送消息到MQ服务端 * 发布（pub）模式 * * @author <a href="mailto:guzhongtao@middol.com">guzhongtao</a> */
 public interface TopicPublisher extends Admin { 
 
 /** * 同步推送消息 * * @param topicMessage 消息对象 * @return TopicMessageSendResult */
 TopicMessageSendResult publish(TopicMessage topicMessage);
 
 /** * 异步推送消息 * * @param topicMessage 消息对象 * @param topicSendCallback 异步结果处理 */
 void publishAsync(TopicMessage topicMessage, TopicSendCallback topicSendCallback);
 
 }

接口具体实现，需要分 阿里云的 和 apache 的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d6dDE5ODgxMTIz_size_16_color_FFFFFF_t_70 1]  
具体实现举一个例子查看， 具体源码请参考最后的 githug地址。

package com.middol.starter.mq.service.impl.aliyun;
 
 import com.aliyun.openservices.ons.api.*;
 import com.middol.starter.mq.exception.TopicMqException;
 import com.middol.starter.mq.pojo.TopicMessage;
 import com.middol.starter.mq.pojo.TopicMessageSendResult;
 import com.middol.starter.mq.service.TopicPublisher;
 import com.middol.starter.mq.service.TopicSendCallback;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /** * 阿里云推送消息到MQ服务端 * 发布（pub）模式 * * @author <a href="mailto:guzhongtao@middol.com">guzhongtao</a> */
 public class AliyunSimpleRocketMqPublisher implements TopicPublisher { 
 
 private Logger logger = LoggerFactory.getLogger(this.getClass());
 /** * 阿里云 rocketmq producer */
 Producer producer;
 
 String beanName;
 
 public AliyunSimpleRocketMqPublisher(Producer producer,String beanName) { 
 this.producer = producer;
 this.beanName = beanName;
 }
 
 public AliyunSimpleRocketMqPublisher() { 
 }
 
 @Override
 public TopicMessageSendResult publish(TopicMessage topicMessage) { 
 Message message = new Message();
 message.setUserProperties(topicMessage.getUserProperties());
 message.setKey(topicMessage.getBussinessKey());
 message.setBody(topicMessage.getMessageBody());
 message.setTag(topicMessage.getTags());
 message.setTopic(topicMessage.getTopicName());
 SendResult sendResult = producer.send(message);
 TopicMessageSendResult topicMessageSendResult = new TopicMessageSendResult();
 topicMessageSendResult.setMessageId(sendResult.getMessageId());
 topicMessageSendResult.setTopicName(sendResult.getTopic());
 return topicMessageSendResult;
 }
 
 @Override
 public void publishAsync(TopicMessage topicMessage, TopicSendCallback topicSendCallback) { 
 Message message = new Message();
 message.setUserProperties(topicMessage.getUserProperties());
 message.setKey(topicMessage.getBussinessKey());
 message.setBody(topicMessage.getMessageBody());
 message.setTag(topicMessage.getTags());
 message.setTopic(topicMessage.getTopicName());
 producer.sendAsync(message, new SendCallback() { 
 @Override
 public void onSuccess(SendResult sendResult) { 
 TopicMessageSendResult topicMessageSendResult = new TopicMessageSendResult();
 topicMessageSendResult.setTopicName(sendResult.getTopic());
 topicMessageSendResult.setMessageId(sendResult.getMessageId());
 topicSendCallback.onSuccess(topicMessageSendResult);
 }
 
 @Override
 public void onException(OnExceptionContext context) { 
 TopicMqException topicMqException = new TopicMqException(context.getException());
 topicMqException.setTopicName(context.getTopic());
 topicMqException.setMessageId(context.getMessageId());
 topicSendCallback.onFail(topicMqException);
 }
 });
 }
 
 @Override
 public boolean isStarted() { 
 return producer.isStarted();
 }
 
 @Override
 public boolean isClosed() { 
 return producer.isClosed();
 }
 
 @Override
 public void start() { 
 logger.info("【MQ】AliyunSimpleRocketMqPublisher["+beanName+"] start...");
 producer.start();
 }
 
 @Override
 public void close() { 
 logger.info("【MQ】AliyunSimpleRocketMqPublisher[" + beanName + "] close...");
 producer.shutdown();
 }
 
 public String getBeanName() { 
 return beanName;
 }
 
 public void setBeanName(String beanName) { 
 this.beanName = beanName;
 }
 
 public Producer getProducer() { 
 return producer;
 }
 
 public void setProducer(Producer producer) { 
 this.producer = producer;
 }
 
 }

package com.middol.starter.mq.service.impl.aliyun;
 
 import com.aliyun.openservices.ons.api.Action;
 import com.aliyun.openservices.ons.api.Consumer;
 import com.middol.starter.mq.pojo.MessageStatus;
 import com.middol.starter.mq.pojo.TopicMessage;
 import com.middol.starter.mq.service.TopicListener;
 import com.middol.starter.mq.service.TopicSubscriber;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /** * 阿里云订阅消息 * * @author <a href="mailto:guzhongtao@middol.com">guzhongtao</a> */
 public class AliyunSimpleRocketMqSubscriber implements TopicSubscriber { 
 
 private Logger logger = LoggerFactory.getLogger(this.getClass());
 
 /** * 阿里云 rocketMq消费服务 */
 Consumer consumer;
 
 String beanName;
 
 public AliyunSimpleRocketMqSubscriber(Consumer consumer, String beanName) { 
 this.consumer = consumer;
 this.beanName = beanName;
 }
 
 public AliyunSimpleRocketMqSubscriber() { 
 }
 
 @Override
 public void subscribe(String topic, String tagExpression, TopicListener listener) { 
 consumer.subscribe(topic, tagExpression, (message, context) -> { 
 TopicMessage topicMessage = new TopicMessage();
 topicMessage.setUserProperties(message.getUserProperties());
 topicMessage.setBussinessKey(message.getKey());
 topicMessage.setMessageBody(message.getBody());
 topicMessage.setTags(message.getTag());
 topicMessage.setTopicName(message.getTopic());
 
 MessageStatus messageStatus = listener.subscribe(topicMessage);
 if (messageStatus.equals(MessageStatus.CommitMessage)) { 
 return Action.CommitMessage;
 } else { 
 return Action.ReconsumeLater;
 }
 });
 }
 
 @Override
 public void unsubscribe(String topicName) { 
 consumer.unsubscribe(topicName);
 }
 
 @Override
 public boolean isStarted() { 
 return consumer.isStarted();
 }
 
 @Override
 public boolean isClosed() { 
 return consumer.isClosed();
 }
 
 @Override
 public void start() { 
 logger.info("【MQ】AliyunSimpleRocketMqSubscriber[" + beanName + "] start...");
 consumer.start();
 }
 
 @Override
 public void close() { 
 logger.info("【MQ】AliyunSimpleRocketMqSubscriber[" + beanName + "] close...");
 consumer.shutdown();
 }
 
 public Consumer getConsumer() { 
 return consumer;
 }
 
 public void setConsumer(Consumer consumer) { 
 this.consumer = consumer;
 }
 
 public String getBeanName() { 
 return beanName;
 }
 
 public void setBeanName(String beanName) { 
 this.beanName = beanName;
 }
 }

这里的架构思想很简单，其实就是在阿里云 或 Apahce rocketmq SDK 的基础上在套一层自己接口的壳子就行了。

## 如何使用 ##

用户在具体自己项目中使用该消息组件时，可根据自己依赖的消息队列产品配置即可：  
如果使用阿里云的rocketmq 可在yml文件中配置如下：

mq:
        aliyun:
          rocketmq:
            enable: true
            access-key-id: xxxxxx
            access-key-secret: bbbbb
            name-server-addr: http://MQ_INST_1666017288766448_BcidZfUM.mq-internet-access.mq-internet.aliyuncs.com:80
            publishers:
              - { beanName: publishService1, groupId: GID_SAAS_DEV, sendMsgTimeoutMillis: 5000}
              - { beanName: xaPublishService1, groupId: GID_XA_SAAS_DEV, sendMsgTimeoutMillis: 5000, messageType: TRANSACTION, checkImmunityTimeInSeconds: 5}
            subscribers:
              - { beanName: subscriberService1, groupId: GID_SAAS_DEV}
              - { beanName: xaSubscriberService1, groupId: GID_XA_SAAS_DEV}

如果你要使用Apache的 RocketMq 则将配置修改如下：

apache:
          rocketmq:
            enable: true
            access-key-id: xxxxx
            access-key-secret: bbbb
            name-server-addr: localhost:9876
            publishers:
              - { beanName: publishService1, groupId: GID_SAAS_DEV, sendMsgTimeout: 5000}
              - { beanName: xaPublishService1, groupId: GID_XA_SAAS_DEV, sendMsgTimeout: 5000, messageType: TRANSACTION}
            subscribers:
              - { beanName: subscriberService1, groupId: GID_SAAS_DEV}
              - { beanName: xaSubscriberService1, groupId: GID_XA_SAAS_DEV}

在 java代码中， 消息发布如下：

package com.middol.mytest.controller;
 
 import cn.hutool.core.util.StrUtil;
 import com.middol.mytest.config.mq.localtransactionexecuter.MyXaTopicLocalTransactionExecuter;
 import com.middol.starter.mq.pojo.TopicMessage;
 import com.middol.starter.mq.pojo.XaTopicMessage;
 import com.middol.starter.mq.service.TopicPublisher;
 import com.middol.starter.mq.service.XaTopicPublisher;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
 import java.nio.charset.StandardCharsets;
 
 /** * 测试 mq * * @author <a href="mailto:guzhongtao@middol.com">guzhongtao</a> */
 @RestController
 @RequestMapping("/api/mq")
 public class MqTestController { 
 
 /** * 普通类型的消息发送服务端 ResourceName 对应yml文件中配置 */
 @Lazy
 @Resource(name = "publishService1")
 private TopicPublisher topicPublisher;
 
 /** * 事务类型的消息发送服务端 ResourceName 对应yml文件中配置 */
 @Lazy
 @Resource(name = "xaPublishService1")
 private XaTopicPublisher xaTopicPublisher;
 
 /** * 普通消息发送测试. * * @param message 消息体 * @return 发送成功 */
 @PostMapping("singlePush")
 public String singlePush(String message) { 
 if (StrUtil.isEmpty(message)) { 
 return "消息体不能为空";
 }
 
 TopicMessage msg = new TopicMessage();
 msg.setTopicName("SAAS-PT");
 msg.setTags("TAG1");
 msg.setBussinessKey(System.currentTimeMillis() + "");
 msg.setMessageBody(message.getBytes(StandardCharsets.UTF_8));
 topicPublisher.publish(msg);
 
 return "发送成功";
 }
 
 /** * 半消息事务类型消息发送 * * @param message 消息体 * @param isCommit 是否要提交测试 true 提交 false 不提交，回滚 * @return 发送成功 */
 @PostMapping("xaPush")
 public String xaPush(String message, Boolean isCommit) { 
 if (StrUtil.isEmpty(message)) { 
 return "消息体不能为空";
 }
 
 XaTopicMessage msg = new XaTopicMessage();
 msg.setLocalTransactionExecuterId(MyXaTopicLocalTransactionExecuter.EXECUTER_ID);
 msg.setTopicName("XA-SAAS");
 msg.setTags("TAG1");
 msg.setBussinessKey(System.currentTimeMillis() + "");
 msg.setMessageBody(message.getBytes(StandardCharsets.UTF_8));
 xaTopicPublisher.publishInTransaction(msg, isCommit);
 
 return "发送成功";
 }
 
 }

消息订阅如下：

package com.middol.mytest.config.mq.listener;
    
    import com.middol.starter.mq.pojo.MessageStatus;
    import com.middol.starter.mq.pojo.TopicMessage;
    import com.middol.starter.mq.service.TopicListener;
    import org.springframework.stereotype.Component;
    
    import java.nio.charset.StandardCharsets;
    
    /** * 消息监听者 * * @author guzt */
    @Component
    public class MyMessageListenerService implements TopicListener { 
        @Override
        public String getSubscriberBeanName() { 
            return "subscriberService1";
        }
    
        @Override
        public String getTopicName() { 
            return "SAAS-PT";
        }
    
        @Override
        public String getTagExpression() { 
            return "*";
        }
    
        @Override
        public MessageStatus subscribe(TopicMessage topicMessage) { 
    
            System.out.println("消费消息 message body = " + new String(topicMessage.getMessageBody(), StandardCharsets.UTF_8));
    
            return MessageStatus.CommitMessage;
        }
    }

上面只是简单说明了一下大致思路，代码只贴出一点，具体源码请移步github，大家一起交流：

**[https://github.com/dwhgygzt/mq-spring-boot-starter][https_github.com_dwhgygzt_mq-spring-boot-starter]**

[https_github.com_dwhgygzt_mq-spring-boot-starter]: https://github.com/dwhgygzt/mq-spring-boot-starter
[https_blog.csdn.net_gzt19881123_article_details_106456362]: https://blog.csdn.net/gzt19881123/article/details/106456362
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d6dDE5ODgxMTIz_size_16_color_FFFFFF_t_70]: /images/20210923/9ba234c7391a40fcb816169d23d01366.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d6dDE5ODgxMTIz_size_16_color_FFFFFF_t_70 1]: /images/20210923/8ebbf9490acf4117a53b5b4435cd935c.png