第一，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处理。所以，小伙伴，你理解了吗？

MQTT是一个轻量级的消息发布/订阅协议，它是实现基于手机客户端的消息推送服务器的理想解决方案。实现MQTT协议的中间件有很多，我用的是Apollo服务器，如何搭建MQTT服务器，请查阅其他资料。这里，主要介绍SpringBoot2.0集成MQTT实现消息推送的功能。好，正式开始：本文采用Gateway绑定的方式，网上也有介绍但不全面，还有其他... <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-integration</artifactId> </dependency>  Spark早期版本中采用Akka作为内部通信部件。  Spark1.3中引入Netty通信框架，为了解决Shuffle的大数据传输问题使用  Spark1.6中Akka和Netty可以配置使用。Netty完全实现了Akka在Spark中的功能。  Spark2系列中，Spark抛弃Akka，使用Netty。 Spark2.x版本使用Netty通讯框架作为内部通讯组件。S @Component @ConfigurationProperties(prefix = "spring. mqtt ") public class Mqtt Properties { private String username = "source"; private String password = "link"; priva... 大体看个样例，描述的是，消息有多种类型，每种类型对应不同的策略，该如何设计呐？首先需要有个抽象类，在类中共用的方法，设计为protected，需要各个类处理的方法，设计为abstract方法，如下假设有登录和Normal 消息要处理，之类改如何设计呐？首先加个@Component注解，在注解中用MessageHandler. 消息类型的方式标明一个类，类似这样，外界只需要给一个消息类型，那么就可以通过 SpringContextUtil.getBean(“MessageHandler” 网上说是因为 client ID 重复，最开始是不相信的，因为我测试只启动了一个客户端。但是却怎么都定位不到异常原因，用重新回到 client ID 重复的这个思路上来：因为程序里同时作为订阅者和. SpringCloud-Finchley.SR1版本中 hystrix-dashboard 报错 /actuator/hystrix.stream 404 Not Found qq_34023259: 本人使用的是SpringBoot是1.x所对应的SpringCloud的版本，所以也是用的第二种方式，验证OK，第一种根本不行。