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

}