1、简述
MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的“轻量级”通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。 MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。
2、创建springboot工程
3、添加依赖
<dependency>
< groupId > org.springframework.bootgroupId >
< artifactId > spring-boot-starter-integration < / artifactId >
< dependency >
< dependency >
< groupId > org.springframework.integration < / groupId >
< artifactId > spring-integration-stream < / artifactId >
< / dependency >
< dependency >
< groupId > org.springframework.integration < / groupId >
< artifactId > spring-integration-mqtt < / artifactId >
< /dependency >
< dependency >
< groupId >org.projectlombok< / groupId >
< artifactId > lombok< / artifactId >
<version>1.16.10</version>
<scope>provided</scope>
< /dependency >
4、MqttInboundProperties
inBound配置属性
package com.chen.config;
import lombok.Data;
/**
* @author: ChenJie
* @date 2018/8/21
*/
@Data
public class MqttInboundProperties {
private Stringurl;
private Stringusername;
private Stringpassword;
private StringclientId;
private Stringtopics;
}
5、MqttOutboundProperties
outBound配置属性
@Setter
@Getter
public class MqttOutboundProperties {
private Stringurls;
private Stringusername;
private Stringpassword;
private StringclientId;
private Stringtopic;
}
6、MqttProperties
配置类
@ConfigurationProperties(prefix ="com.mqtt")
public class MqttProperties {
private MqttInboundPropertiesinbound;
private MqttOutboundPropertiesoutbound;
public MqttInboundPropertiesgetInbound() {
return inbound;
}
public void setInbound(MqttInboundProperties inbound) {
this.inbound = inbound;
}
public MqttOutboundPropertiesgetOutbound() {
return outbound;
}
public void setOutbound(MqttOutboundProperties outbound) {
this.outbound = outbound;
}
}
7、MqttInboundConfiguration
消息接收处理类
@Configuration
@Slf4j
public class MqttInboundConfiguration {
@Autowired
private MqttPropertiesmqttProperties;
@Bean
public MessageChannelmqttInputChannel() {
return new DirectChannel();
}
@Bean
public MessageProducerinbound(MqttPahoClientFactory mqttPahoClientFactory) {
String[] inboundTopics =mqttProperties.getInbound().getTopics().split(",");
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getInbound().getUrl(), mqttProperties.getInbound().getClientId(), mqttPahoClientFactory,inboundTopics);
adapter.setCompletionTimeout(5000);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}
@Bean
@ServiceActivator(inputChannel ="mqttInputChannel")
public MessageHandlerhandler() {
return new MessageHandler() {
@Override
public void handleMessage(Message message)throws MessagingException {
log.info("收到消息："+(String) message.getPayload());
}
};
}
}
8、MqttOutboundConfiguration
消息发送配置类
@Configuration
public class MqttOutboundConfiguration {
@Autowired
private MqttPropertiesmqttProperties;
@Bean
public MqttPahoClientFactorymqttClientFactory() {
String[] serverURIs =mqttProperties.getOutbound().getUrls().split(",");
DefaultMqttPahoClientFactory factory =new DefaultMqttPahoClientFactory();
factory.setServerURIs(serverURIs);
factory.setCleanSession(false);
factory.setUserName(mqttProperties.getOutbound().getUsername());
factory.setPassword(mqttProperties.getOutbound().getPassword());
return factory;
}
@Bean
@ServiceActivator(inputChannel ="mqttOutboundChannel")
public MessageHandlermqttOutbound() {
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttProperties.getOutbound().getClientId(),
mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(mqttProperties.getOutbound().getTopic());
return messageHandler;
}
@Bean
public MessageChannelmqttOutboundChannel() {
return new DirectChannel();
}
}
9、MqttGateway
消息发送service，可以直接调用来发送消息
@Component
@MessagingGateway(defaultRequestChannel ="mqttOutboundChannel")
public interface MqttGateway {
void sendToMqtt(String data);
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS)int qos, String payload);
}
10、http接口触发mqtt消息分发， MessageController
@RestController
public class MessageController {
@Autowired
MqttGatewaymqttGateway;
@RequestMapping(value="/sendMsg")
public StringsendMsg(@RequestParam String message){
//调用网关接口发送消息
mqttGateway.sendToMqtt(message);
return "success";
}
}
11、springboot启动类
@SpringBootApplication
@Configuration
@EnableConfigurationProperties(MqttProperties.class)
public class MqttSpringbootApplication {
public static void main(String[] args) {
SpringApplication.run(MqttSpringbootApplication.class, args);
}
}
