springboot Mqtt消息订阅和发布
行云的逆袭 人气:21.添加依赖
<dependency> <groupId>org.jetlinks</groupId> <artifactId>netty-mqtt-client</artifactId> <version>1.0.0</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.13.2</version> <scope>test</scope> </dependency>
2.源码
application.yml
#mqtt配置 mqtt: username: admin password: 123456 #推送信息的连接地址 url: localhost port: 1884 #默认发送的主题 defaultTopic: topic #clientid clientId: client #连接超时时间 单位为秒 completionTimeout: 300 #设置会话心跳时间 单位为秒 keepAliveInterval: 20
MqttProperties.java
import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; @Data @ConfigurationProperties(prefix = "mqtt") public class MqttProperties { private String username; private String password; private String url; private int port; private String clientId; private String defaultTopic; private int completionTimeout; private int keepAliveInterval; }
MqttConfig.java
import com.xingyun.netty.mqtt.prop.MqttProperties; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import lombok.AllArgsConstructor; import org.jetlinks.mqtt.client.*; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @AllArgsConstructor @Configuration @EnableConfigurationProperties(MqttProperties.class) public class MqttConfig { private final MqttProperties mqttProperties; @Bean public MqttClientConfig getMqttClientConfig() { MqttClientConfig mqttClientConfig = new MqttClientConfig(); mqttClientConfig.setClientId(mqttProperties.getClientId()); mqttClientConfig.setUsername(mqttProperties.getClientId()); mqttClientConfig.setPassword(mqttProperties.getPassword()); /*mqttClientConfig.setTimeoutSeconds(mqttProperties.getCompletionTimeout()); mqttClientConfig.setRetryInterval(mqttProperties.getKeepAliveInterval()); mqttClientConfig.setProtocolVersion(MqttVersion.MQTT_3_1_1); mqttClientConfig.setReconnect(true);*/ return mqttClientConfig; } @Bean public MqttClient getMqttClient(){ EventLoopGroup loop = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2); MqttClient mqttClient = new MqttClientImpl(getMqttClientConfig(),null); mqttClient.setEventLoop(loop); mqttClient.setCallback(getMqttClientCallback()); mqttClient.connect(mqttProperties.getUrl(), mqttProperties.getPort()).addListener(future -> { if (future.isSuccess()){ System.out.println("mqtt客户端已建立连接"); //#为多层通配符,+为单层通配符 mqttClient.on("#",getMqttHandler()); } }); return mqttClient; } @Bean public MqttHandler getMqttHandler(){ return (topic,payload) -> { System.out.println("消息主题:" + topic); System.out.println("消息内容:" + payload); }; } @Bean public MqttClientCallback getMqttClientCallback(){ return new MqttClientCallback() { @Override public void connectionLost(Throwable cause) { cause.printStackTrace(); } @Override public void onSuccessfulReconnect() { System.out.println("客户端已重连"); } }; } }
3.运行测试
客户端利用不同主题,发送消息
控制台
消息主题:testTopic/001
消息内容:PooledSlicedByteBuf(ridx: 0, widx: 15, cap: 15/15, unwrapped: PooledUnsafeDirectByteBuf(ridx: 32, widx: 32, cap: 512))
消息主题:testTopic/001
消息内容:PooledSlicedByteBuf(ridx: 0, widx: 15, cap: 15/15, unwrapped: PooledUnsafeDirectByteBuf(ridx: 32, widx: 32, cap: 512))
消息主题:test/sub/001
消息内容:PooledSlicedByteBuf(ridx: 0, widx: 15, cap: 15/15, unwrapped: PooledUnsafeDirectByteBuf(ridx: 31, widx: 31, cap: 496))
消息主题:test1
消息内容:PooledSlicedByteBuf(ridx: 0, widx: 15, cap: 15/15, unwrapped: PooledUnsafeDirectByteBuf(ridx: 24, widx: 24, cap: 496))
单元测试发布消息
MqttSeviceDemo.java
import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import org.jetlinks.mqtt.client.MqttClient; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest public class MqttSeviceDemo { @Autowired private MqttClient mqttClient; @Test public void publishMessage(){ String test = "I am client9527"; byte[] bytes = test.getBytes(); ByteBuf byteBuf = Unpooled.copiedBuffer(bytes); mqttClient.publish("test/pub/001",byteBuf); System.out.println("消息已发布"); } }
客户端订阅到消息
到此这篇关于springboot整合netty-mqtt-client实现Mqtt消息的订阅和发布示例的文章就介绍到这了,更多相关springboot Mqtt消息订阅和发布内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持!
加载全部内容