Springboot如何详细实现mqtt客户端集成？

本文共计816个文字，预计阅读时间需要4分钟。

目录

1.前言

2.引入依赖

3.配置文件

4.MQTT消息类型

5.MQTT消息接收器

6.MQTT消息发送器

7.测试MQTT消息发送

1.前言

在这里，我们使用Spring Boot搭建一个轻量级的MQTT客户端，连接到MQTT的Broker服务。

目录

• 1. 前言
• 2. 引入依赖
• 3. 配置文件
• 4. MQTT消息类
• 5. MQTT消息接收器
• 6. MQTT消息发送器
• 7. 测试MQTT发送消息

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内容请搜索自由互联以前的文章或继续浏览下面的相关文章希望大家以后多多支持自由互联！