SpringBoot整合阿里云消息队列MQ

一、概述

      消息队列&＃xff08;Message Queue&＃xff0c;简称 MQ&＃xff09;是阿里巴巴集团中间件技术部自主研发的专业消息中间件。产品基于高可用分布式集群技术&＃xff0c;提供消息发布订阅、消息轨迹查询、定时&＃xff08;延时&＃xff09;消息、资源统计、监控报警等一系列消息云服务&＃xff0c;是企业级互联网架构的核心产品。MQ 历史超过9年&＃xff0c;为分布式应用系统提供异步解耦、削峰填谷的能力&＃xff0c;同时具备海量消息堆积、高吞吐、可靠重试等互联网应用所需的特性&＃xff0c;是阿里巴巴双11使用的核心产品。

二、Spring Boot集成

      阿里云官方文档及demo是整合Spring&＃xff0c;使用的是xml配置&＃xff0c;在这里&＃xff0c;我结合官方文档&＃xff0c;在Spring Boot框架下整合MQ,由于在网上相关文档及demo不是很多&＃xff0c;所以有很多不足的地方&＃xff0c;如果有更好的建议&＃xff0c;还请多多指教。
      基础的项目搭建在这里就不多说了&＃xff0c;在网上也有很多教程&＃xff0c;就直接进入正题了。

1. 参数的配置

      我把MQ相关的参数写在Application.properties中

#aliyun MQ config
producerId&＃61;*** //替换为自己的账号
consumerId&＃61;***
accessKey&＃61;***
secretKey&＃61;***
tag&＃61;***
topic&＃61;***
onsAddr&＃61;http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet
sendMsgTimeoutMillis&＃61;3000
suspendTimeMillis&＃61;100
maxReconsumeTimes&＃61;20

2. demo

&＃64;Configuration标注在类上&＃xff0c;相当于把该类作为spring的xml配置文件中的&＃xff0c;作用为&＃xff1a;配置spring容器(应用上下文)
&＃64;Bean标注在方法上(返回某个实例的方法)&＃xff0c;等价于spring的xml配置文件中的&＃xff0c;作用为&＃xff1a;注册bean对象

&＃64;Configuration
public class AliMQConfig {&＃64;Value("${producerId}")public String producerId;&＃64;Value("${consumerId}")public String consumerId;&＃64;Value("${accessKey}")public String accessKey;&＃64;Value("${secretKey}")public String secretKey;&＃64;Value("${topic}")public String topic;&＃64;Value("${tag}")public String tag;&＃64;Value("${onsAddr}")public String onsAddr;//超时时间&＃64;Value("${sendMsgTimeoutMillis}")public String sendMsgTimeoutMillis;&＃64;Value("${suspendTimeMillis}")public String suspendTimeMillis;&＃64;Value("${maxReconsumeTimes}")public String maxReconsumeTimes;&＃64;Bean(initMethod &＃61; "start", destroyMethod &＃61; "shutdown")public ProducerBean getProducer() {ProducerBean producerBean &＃61; new ProducerBean();Properties properties &＃61; new Properties();properties.put(PropertyKeyConst.ProducerId, producerId);// AccessKey 阿里云身份验证&＃xff0c;在阿里云服务器管理控制台创建properties.put(PropertyKeyConst.AccessKey, accessKey);// SecretKey 阿里云身份验证&＃xff0c;在阿里云服务器管理控制台创建properties.put(PropertyKeyConst.SecretKey, secretKey);properties.put(PropertyKeyConst.SendMsgTimeoutMillis, sendMsgTimeoutMillis);properties.put(PropertyKeyConst.ONSAddr, onsAddr);producerBean.setProperties(properties);return producerBean;}&＃64;Bean(initMethod &＃61; "start", destroyMethod &＃61; "shutdown")public ConsumerBean getConsumer() {ConsumerBean consumerBean &＃61; new ConsumerBean();Properties properties &＃61; new Properties();properties.put(PropertyKeyConst.ConsumerId, consumerId);// AccessKey 阿里云身份验证&＃xff0c;在阿里云服务器管理控制台创建properties.put(PropertyKeyConst.AccessKey, accessKey);// SecretKey 阿里云身份验证&＃xff0c;在阿里云服务器管理控制台创建properties.put(PropertyKeyConst.SecretKey, secretKey);properties.put(PropertyKeyConst.SuspendTimeMillis, suspendTimeMillis);properties.put(PropertyKeyConst.MaxReconsumeTimes, maxReconsumeTimes);properties.put(PropertyKeyConst.ONSAddr, onsAddr);consumerBean.setProperties(properties);Subscription subscription &＃61; new Subscription();subscription.setTopic(topic);subscription.setExpression(tag);Map map &＃61; new HashMap();map.put(subscription, new AliMQConsumerListener());consumerBean.setSubscriptionTable(map);return consumerBean;}

这个类相当于官方demo里的xml配置&＃xff0c;由于SpringBoot旨在减少xml配置&＃xff0c;所以在这里就把xml配置转换为java代码。
下面是生产者发送消息&＃xff1a;

&＃64;Component
public class AliMQUtil {private static final Logger logger &＃61; LoggerFactory.getLogger(AliMQUtil.class);&＃64;Autowiredprivate AliMQConfig aliMQConfig;&＃64;Value("${topic}")public String topic;// 发送消息public void sendMessage(String tag, String key, byte[] body) {Producer producer &＃61; aliMQConfig.getProducer();Message msg &＃61; new Message(topic, tag, body);//msg.setKey(key);try {SendResult sendResult &＃61; producer.send(msg);if (sendResult !&＃61; null) {logger.info("消息发送成功&＃xff1a;" &＃43; sendResult.toString());}} catch (ONSClientException e) {logger.info("消息发送失败&＃xff1a;", e);// 出现异常意味着发送失败&＃xff0c;为了避免消息丢失&＃xff0c;建议缓存该消息然后进行重试。}}
}

消费者监听&＃xff1a;

public class AliMQConsumerListener implements MessageListener {private static final Logger logger &＃61; LoggerFactory.getLogger(AliMQConsumerListener.class);&＃64;Overridepublic Action consume(Message message, ConsumeContext context) {String msg &＃61; "";try {//do something..msg &＃61; new String(message.getBody(), "UTF-8");logger.info("订阅消息&＃xff1a;" &＃43; msg);return Action.CommitMessage;} catch (Exception e) {//消费失败logger.info("消费失败&＃xff1a;" &＃43; msg);return Action.ReconsumeLater;}}

当项目启动&＃xff0c;消费者就开始监听消息&＃xff0c;实测&＃xff0c;项目启动后台就会打印消息日志。
由于官方文档中说明MQ 的消费者和生产者客户端对象是线程安全的&＃xff0c;可以在多个线程之间共享使用&＃xff0c;所以这里配置的是跟随项目启动&＃xff0c;创建一个实例&＃xff0c;多线程之间可以共享这一实例。由于对多线程的理解不是很深刻&＃xff0c;如果不当&＃xff0c;还请指正。
      以上就是我在Spring Boot中对于MQ的整合方案&＃xff0c;有很多不足的地方&＃xff0c;希望和大家一起交流学习。

补充&＃xff1a;MQTT的使用

针对用户在移动互联网以及物联网领域的存在的特殊消息传输需求&＃xff0c;MQ 开放了 MQTT 协议的完整支持。

消息队列遥测传输&＃xff08;Message Queuing Telemetry Transport&＃xff0c;简称 MQTT&＃xff09;是一种轻量的&＃xff0c;基于发布订阅模型的即时通讯协议。该协议设计开放&＃xff0c;协议简单&＃xff0c;平台支持丰富&＃xff0c;几乎可以把所有联网物品和外部连接起来&＃xff0c;因此在移动互联网和物联网领域拥有众多优势。

协议的特点包括&＃xff1a;

使用发布/订阅消息模式&＃xff0c;提供一对多的消息分发&＃xff0c;解除了应用程序之间的耦合&＃xff1b;
对负载内容屏蔽的消息传输&＃xff1b;
使用 TCP/IP 提供基础的网络连接&＃xff1b;
有三种级别的消息传递服务&＃xff1b;
小型传输&＃xff0c;开销很小&＃xff08;固定长度的头部是2字节&＃xff09;&＃xff0c;协议交换最小化&＃xff0c;以降低网络流量。

客户端分布&＃xff1a;
使用 MQTT 协议的客户端作为移动端接入 MQ&＃xff0c;一般分布在公网环境&＃xff0c;比如嵌入式设备、移动手机、平板、浏览器之类的平台上。
使用 MQ 协议的客户端一般作为业务上的服务端接入 MQ&＃xff0c;应该部署在阿里云的 ECS 环境。

MQ服务端&＃xff1a;MQ发送MQTT消息&＃xff0c;只需要在原来的代码做点小改动即可。

&＃64;Component
public class AliMQUtil {private static final Logger logger &＃61; LoggerFactory.getLogger(AliMQUtil.class);&＃64;Autowiredprivate AliMQConfig aliMQConfig;private static String topic;private static String groupId;&＃64;Value("${topic}")public String top;&＃64;Value("${groupId}")public String gId;public void setAliMQConfig(AliMQConfig aliMQConfig) {this.aliMQConfig &＃61; aliMQConfig;}public AliMQConfig getAliMQConfig() {return aliMQConfig;}private static AliMQUtil aliMQUtil;&＃64;PostConstructpublic void init() {topic &＃61; top;groupId &＃61; gId;aliMQUtil &＃61; this;aliMQUtil.aliMQConfig &＃61; this.aliMQConfig;}// 发送消息public static void sendMessage(String vehicleId, byte[] body) {Producer producer &＃61; aliMQUtil.aliMQConfig.getProducer();final Message msg &＃61; new Message(topic, // MQ 消息的 Topic&＃xff0c;需要事先创建"MQ2MQTT", // MQ Tag&＃xff0c;通过 MQ 向 MQTT 客户端发消息时&＃xff0c;必须指定 MQ2MQTT 作为 Tag&＃xff0c;其他 Tag 或者不设都将导致 MQTT 客户端收不到消息body);// 消息体&＃xff0c;和 MQTT 的 body 对应/*** 使用 MQ 客户端给 MQTT 设备发送 P2P 消息时&＃xff0c;需要在 MQ 消息中设置 mqttSecondTopic 属性* 设置的值是“/p2p/”&＃43;目标 ClientID*/String targetClientID &＃61; groupId &＃43; "&＃64;&＃64;&＃64;" &＃43; vehicleId;msg.putUserProperties("mqttSecondTopic", "/p2p/" &＃43; targetClientID);try {SendResult sendResult &＃61; producer.send(msg);if (sendResult !&＃61; null) {logger.info("消息发送成功&＃xff1a;" &＃43; sendResult.toString());}} catch (ONSClientException e) {logger.info("消息发送失败&＃xff1a;", e);// 出现异常意味着发送失败&＃xff0c;为了避免消息丢失&＃xff0c;建议缓存该消息然后进行重试。}}
}


客户端&＃xff1a;不能开发语言的demo都可以在阿里云帮助文档找到&＃xff0c;我这里使用的是java&＃xff0c;用于简单测试。

public static void acceptMessage() throws Exception {/*** 设置当前用户私有的 MQTT 的接入点。例如此处示意使用 XXX&＃xff0c;实际使用请替换用户自己的接入点。接入点的获取方法是&＃xff0c;在控制台创建 MQTT* 实例&＃xff0c;每个实例都会分配一个接入点域名。*/final String broker &＃61; "tcp://****.mqtt.aliyuncs.com:1883";/*** 设置阿里云的 AccessKey&＃xff0c;用于鉴权*/final String acessKey &＃61; "***";/*** 设置阿里云的 SecretKey&＃xff0c;用于鉴权*/final String secretKey &＃61; "***";/*** 发消息使用的一级 Topic&＃xff0c;需要先在 MQ 控制台里创建*/final String topic &＃61; "***";/*** MQTT 的 ClientID&＃xff0c;一般由两部分组成&＃xff0c;GroupID&＃64;&＃64;&＃64;DeviceID* 其中 GroupID 在 MQ 控制台里创建* DeviceID 由应用方设置&＃xff0c;可能是设备编号等&＃xff0c;需要唯一&＃xff0c;否则服务端拒绝重复的 ClientID 连接*/final String clientId &＃61; "***&＃64;&＃64;&＃64;0001";String sign;MemoryPersistence persistence &＃61; new MemoryPersistence();try {final MqttClient sampleClient &＃61; new MqttClient(broker, clientId, persistence);final MqttConnectOptions connOpts &＃61; new MqttConnectOptions();System.out.println("Connecting to broker: " &＃43; broker);/*** 计算签名&＃xff0c;将签名作为 MQTT 的 password* 签名的计算方法&＃xff0c;参考工具类 MacSignature&＃xff0c;第一个参数是 ClientID 的前半部分&＃xff0c;即 GroupID* 第二个参数阿里云的 SecretKey*/sign &＃61; MacSignature.macSignature(clientId.split("&＃64;&＃64;&＃64;")[0], secretKey);/*** 设置订阅方订阅的 Topic 集合&＃xff0c;此处遵循 MQTT 的订阅规则&＃xff0c;可以是一级 Topic&＃xff0c;二级 Topic&＃xff0c;P2P 消息请订阅/p2p*/final String[] topicFilters &＃61; new String[] { topic &＃43; "/notice/", topic &＃43; "/p2p" };final int[] qos &＃61; { 0, 0 };connOpts.setUserName(acessKey);connOpts.setServerURIs(new String[] { broker });connOpts.setPassword(sign.toCharArray());connOpts.setCleanSession(true);connOpts.setKeepAliveInterval(90);connOpts.setAutomaticReconnect(true);sampleClient.setCallback(new MqttCallbackExtended() {public void connectComplete(boolean reconnect, String serverURI) {System.out.println("connect success");// 连接成功&＃xff0c;需要上传客户端所有的订阅关系try {sampleClient.subscribe(topicFilters, qos);} catch (MqttException e) {e.printStackTrace();}}public void connectionLost(Throwable throwable) {System.out.println("mqtt connection lost");}public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {System.out.println("messageArrived:" &＃43; topic &＃43; "------" &＃43; new String(mqttMessage.getPayload()));}public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {System.out.println("deliveryComplete:" &＃43; iMqttDeliveryToken.getMessageId());}});// 客户端每次上线都必须上传自己所有涉及的订阅关系&＃xff0c;否则可能会导致消息接收延迟sampleClient.connect(connOpts);// 每个客户端最多允许存在30个订阅关系&＃xff0c;超出限制可能会丢弃导致收不到部分消息sampleClient.subscribe(topicFilters, qos);Thread.sleep(Integer.MAX_VALUE);} catch (Exception me) {me.printStackTrace();}}