springboot整合netty-mqtt-client实现Mqtt消息的订阅和发布示例

< 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 >

#mqtt配置

mqtt:

   username: admin

   password: 123456

   #推送信息的连接地址

   url: localhost

   port: 1884

   #默认发送的主题

   defaultTopic: topic

   #clientid

   clientId: client

   #连接超时时间 单位为秒

   completionTimeout: 300

   #设置会话心跳时间 单位为秒

   keepAliveInterval: 20

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;

}

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( "客户端已重连" );

             }

         };

     }

}

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( "消息已发布" );

     }

}