1. 前言
这里我们使用springboot搭建一个轻量级的mqtt客户端，连接mqtt的Broker服务。
连接信息写在配置文件里application.properties
spring.mqtt.username=admin
spring.mqtt.mqpassword=admin
spring.mqtt.host-url= tcp://127.0.0.1:1883
spring.mqtt.client-id= server_client_${random.value}
spring.mqtt.default-topic= $SYS/brokers/ /clients/#
spring.mqtt.completionTimeout= 3000
spring.mqtt.keepAlive= 60
2. 引入依赖
<!--mqtt -->
<dependency>
<groupId>org.springframework.boot</groupId>
<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>
3. 配置文件
新建MqttProperties.java文件，初始化application里的mqtt配置项
@ConfigurationProperties("spring.mqtt")
@Component
@Getter
@Setter
public class MqttProperties {
private String username;
private String mqpassword;
private String hostUrl;
private String clientId;
private String defaultTopic;
private String completionTimeout;
private Integer keepAlive;
} 新建MqttConfiguration.java文件，为mqtt做初始化配置
@Configuration
@Slf4j
public class MqttConfiguration {
@Autowired
private MqttProperties mqttProperties;
/**
* 事件触发
*/
@Autowired
private ApplicationEventPublisher eventPublisher;
@Bean
public MqttConnectOptions getMqttConnectOptions(){
MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
mqttConnectOptions.setUserName(mqttProperties.getUsername());
mqttConnectOptions.setPassword(mqttProperties.getMqpassword().toCharArray());
mqttConnectOptions.setServerURIs(new String[]{mqttProperties.getHostUrl()});
mqttConnectOptions.setKeepAliveInterval(2);
mqttConnectOptions.setKeepAliveInterval(mqttProperties.getKeepAlive());
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(mqttProperties.getClientId() "_inbound", mqttClientFactory(),
mqttProperties.getDefaultTopic().split(","));
adapter.setCompletionTimeout(Long.valueOf(mqttProperties.getCompletionTimeout()));
adapter.setConverter(new DefaultPahoMessageConverter());
//默认添加TopicName中所有tipic
adapter.addTopic(" / /test");
adapter.setQos(2);
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 qos = message.getHeaders().get("mqtt_receivedQos").toString();
//触发事件 这里不再做业务处理，包 listener中做处理
eventPublisher.publishEvent(new MqttEvent(this,topic,message.getPayload().toString()));
}
};
}
/**
* 发送消息和消费消息Channel可以使用相同MqttPahoClientFactory
*
* @return
*/
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
// 在这里进行mqttOutboundChannel的相关设置
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttProperties.getClientId(), mqttClientFactory());
// 如果设置成true，发送消息时将不会阻塞。
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(mqttProperties.getDefaultTopic());
return messageHandler;
}
@Bean
public MessageChannel mqttOutboundChannel() {
return new DirectChannel();
}
}4. MQTT消息类
新建MqttEvent.java 消息类。用于发送mqtt的消息
@Getter
public class MqttEvent extends ApplicationEvent {
private String topic;
/**
* 发送的消息
*/
private String message;
public MqttEvent(Object source,String topic,String message) {
super(source);
this.topic = topic;
this.message = message;
}
}5. MQTT消息接收器
新建JobListener.java文件作为 mqtt的消息接收类
@Slf4j
@Component
public class JobListener {
@Autowired
DeviceDao deviceDao;
/**
* 监听topic
* @param mqttEvent
*/
@EventListener(condition = "#mqttEvent.topic.startsWith('pay')")
public void onEmqttCall1(MqttEvent mqttEvent) throws Exception {
String topic = mqttEvent.getTopic();
//写逻辑处理
}
/**
* 监听topic
* @param mqttEvent
*/
@EventListener(condition = "#mqttEvent.topic.equals('device')")
public void onEmqttCallT(MqttEvent mqttEvent){
log.info("接收到消11111111111：" mqttEvent.getMessage());
}
}6. MQTT消息发送器
新建MqttGateway.java 提供发送mqttt消息的接口服务
@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);
}7. 测试MQTT发送消息
@SpringBootTest
public class Test3 {
@Autowired
MqttGateway mqttGateway;
@Test
public void mqttTest () {
mqttGateway.sendToMqtt("111//222/33","消息内容");
}
}到此这篇关于Springboot集成mqtt客户端详解的文章就介绍到这了,更多相关Springboot集成mqtt内容请搜索Devmax以前的文章或继续浏览下面的相关文章希望大家以后多多支持Devmax！