SpringBoot实践之---集成MQTT消息推送和消息订阅功能

MQTT是一个轻量级的消息发布/订阅协议，它是实现基于手机客户端的消息推送服务器的理想解决方案。实现MQTT协议的中间件有很多，我用的是Apollo服务器，如何搭建MQTT服务器，请查阅其他资料。这里，主要介绍SpringBoot2.0集成MQTT实现消息推送的功能。好，正式开始：本文采用Gateway绑定的方式，网上也有介绍但不全面，还有其他...

luckykapok918

22834人浏览 · 2018-12-11 13:44:03

luckykapok918 · 2018-12-11 13:44:03 发布

MQTT是一个轻量级的消息发布/订阅协议，它是实现基于手机客户端的消息推送服务器的理想解决方案。

实现MQTT协议的中间件有很多，我用的是Apollo服务器，如何搭建MQTT服务器，请查阅其他资料。这里，主要介绍SpringBoot2.0集成MQTT实现消息推送的功能。好，正式开始：

本文采用Gateway绑定的方式，网上也有介绍但不全面，还有其他采用Paho MQTT Client库的方式实现的。

-------------------------------------------------------------------------消息推送--------------------------------------------------------------------------------------

第一，pom配置，我们需要引入相关jar：

<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>

第二，配置MQTT服务器基本信息，

在springBoot配置文件application.properties中配置，添加如下：

#MQTT配置信息
#MQTT-用户名
spring.mqtt.username=admin
#MQTT-密码
spring.mqtt.password=password
#MQTT-服务器连接地址，如果有多个，用逗号隔开，如：tcp://127.0.0.1:61613，tcp://192.168.2.133:61613
spring.mqtt.url=tcp://127.0.0.1:61613
#MQTT-连接服务器默认客户端ID
spring.mqtt.client.id=mqttId
#MQTT-默认的消息推送主题，实际可在调用接口时指定
spring.mqtt.default.topic=topic

第三，配置MQTT消息推送配置类

/**
* 〈一句话功能简述〉<br>
* 〈MQTT发送消息配置〉
*
* @author AnswerChang
* @create 2018/6/4
* @since 1.0.0
*/
@Configuration
@IntegrationComponentScan
public class MqttSenderConfig {

@Value("${spring.mqtt.username}")
private String username;

@Value("${spring.mqtt.password}")
private String password;

@Value("${spring.mqtt.url}")
private String hostUrl;

@Value("${spring.mqtt.client.id}")
private String clientId;

@Value("${spring.mqtt.default.topic}")
private String defaultTopic;

@Bean
public MqttConnectOptions getMqttConnectOptions(){
MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
mqttConnectOptions.setUserName(username);
mqttConnectOptions.setPassword(password.toCharArray());
mqttConnectOptions.setServerURIs(new String[]{hostUrl});
mqttConnectOptions.setKeepAliveInterval(2);
return mqttConnectOptions;
}
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(getMqttConnectOptions());
return factory;
}
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(defaultTopic);
return messageHandler;
}
@Bean
public MessageChannel mqttOutboundChannel() {
return new DirectChannel();
}
}

第四，配置MqttGateway消息推送接口类

在sendToMqtt(String data,@Header(MqttHeaders.TOPIC)String topic)接口中，data为发送的消息内容，topic为主题。指定topic，则我们的接口可以根据需要，向不同的主题发送消息，方便灵活应用。如果不指定，则使用默认配置的主题。

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
void sendToMqtt(String data,@Header(MqttHeaders.TOPIC) String topic);
}
最后，写个接口类测试下功能，用Postman调用sendMqtt.do接口，往hello主题发送消息，用MQTTLens订阅hello主题，从下面截图，可以看出可以正常往MQTT服务发送消息了，而且可以订阅到。

@RestController
@RequestMapping("/test")
public class TestController {

@Autowired
private MqttGateway mqttGateway;

@RequestMapping("/sendMqtt.do")
public String sendMqtt(String sendData){
mqttGateway.sendToMqtt(sendData,"hello");
return "OK";
}
}

-------------------------------------------------------------------------消息订阅--------------------------------------------------------------------------------------

第一，pom配置，引入相关jar:

第二，配置MQTT服务器基本信息

在springBoot配置文件application.properties中配置，添加如下：

第三，配置MQTT消息接收处理类：

/**
* 〈一句话功能简述〉<br>
* 〈MQTT接收消息处理〉
*
* @author lenovo
* @create 2018/6/4
* @since 1.0.0
*/
@Configuration
@IntegrationComponentScan
public class MqttReceiveConfig {

@Value("${spring.mqtt.username}")
private String username;

@Value("${spring.mqtt.password}")
private String password;

@Value("${spring.mqtt.url}")
private String hostUrl;

@Value("${spring.mqtt.client.id}")
private String clientId;

@Value("${spring.mqtt.default.topic}")
private String defaultTopic;

@Value("${spring.mqtt.completionTimeout}")
private int completionTimeout ; //连接超时

@Bean
public MqttConnectOptions getMqttConnectOptions(){
MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
mqttConnectOptions.setUserName(username);
mqttConnectOptions.setPassword(password.toCharArray());
mqttConnectOptions.setServerURIs(new String[]{hostUrl});
mqttConnectOptions.setKeepAliveInterval(2);
return mqttConnectOptions;
}
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(getMqttConnectOptions());
return factory;
}

//接收通道
@Bean
public MessageChannel mqttInputChannel() {
return new DirectChannel();
}

//配置client,监听的topic
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(clientId+"_inbound", mqttClientFactory(),
"hello","hello1");
adapter.setCompletionTimeout(completionTimeout);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}

//通过通道获取数据
@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
String type = topic.substring(topic.lastIndexOf("/")+1, topic.length());
if("hello".equalsIgnoreCase(topic)){
System.out.println("hello,fuckXX,"+message.getPayload().toString());
}else if("hello1".equalsIgnoreCase(topic)){
System.out.println("hello1,fuckXX,"+message.getPayload().toString());
}
}
};
}
}

第四，启动服务测试

使用postment调用上一篇的MQTT发送接口，分别往hello,hello1两个topic发送消息，测试接收情况：

由此看出，可以正常监听topic并接收处理消息了。

看到这里，朋友们可能有疑问，如果我要配置多个client，应该怎么处理呢？这个也简单，我们只要配置多个通道即可，简单代码如下：

//通道2
@Bean
public MessageChannel mqttInputChannelTwo() {
return new DirectChannel();
}
//配置client2，监听的topic:hell2,hello3
@Bean
public MessageProducer inbound1() {
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(clientId+"_inboundTwo", mqttClientFactory(),
"hello2","hello3");
adapter.setCompletionTimeout(completionTimeout);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInputChannelTwo());
return adapter;
}

//通过通道2获取数据
@Bean
@ServiceActivator(inputChannel = "mqttInputChannelTwo")
public MessageHandler handlerTwo() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
String type = topic.substring(topic.lastIndexOf("/")+1, topic.length());
if("hello2".equalsIgnoreCase(topic)){
System.out.println("hello2 clientTwo,"+message.getPayload().toString());
}else if("hello3".equalsIgnoreCase(topic)){
System.out.println("hello3 clientTwo,"+message.getPayload().toString());
}
}
};
}
这样一来，我们就配置了两个client，client1监听处理hello、hello1主题消息，client2监听处理hello2、hello3主题，测试一下：

从输出结果可以看出，我们发送不同的消息，分别由不同的client处理。所以，小伙伴，你理解了吗？

向您推荐>>Eolink开发者社区

权威｜前沿｜技术｜干货｜国内首个API全生命周期开发者社区

更多推荐

沃云统一开发平台介绍

沃云集成平台研发平台介绍1.平台优势2.平台原理3.研发平台使用方法4.遇到的问题5.现阶段实现的功能6.后续需要补充的功能和优化内容研发平台介绍1.平台优势解决孤岛式应用，实现能力共享；现有系统框架过于复杂，跨系统业务处理成本居高不下，协同服务共享，降低运维成本；提高项目应用资源监控能力，改善资源利用率；业务微服务化，快速发布、快速部署，快速响应业务需求变化；沃云平台不仅提供了自动化的、可快速部

云原生

(20200916 Solved)docker-compose up创建容器自动退出

问题描述如题，创建容器后自动退出了。并且docker start container无效解决方案原因是缺失了控制终端的配置，需要在docker-compose.yml中增加tty:true ，有时候这样也不行，需要再增加一个command:/bin/bash，命令不一定是这个，需要是一个不会退出的命令，然后用-d后台启动容器。Referencesdocker-compose启动容器后自动退出...

云原生

基于docker的test-containers环境百宝箱

笔者语录：我开了个公众号【Java你我他】，欢迎大家关注。在很多时候，程序猿们更关注代码本身，而不愿意把时间花费在环境搭建上，这也是Docker变得越来越受欢迎的原因之一。test-containe是Docker生态圈中的一颗新星，其主要针对测试领域、背靠Docker实现环境百宝箱功能。 test-containers：你要的环境，我都有~ 假设我们现在需要一个redis-clust