MQTT订阅和发送实战(JAVA)

本列子使用MQTT+EMQX+Springboot
1.在pom.xml中引入依赖

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

2.在application.yml中添加配置

mqtt:
  username: 用户名
  password: 密码
  # 推送信息的连接地址，如果有多个，用逗号隔开，如：tcp://127.0.0.1:61613,tcp://192.168.1.61:61613
  url: tcp://127.0.0.1:1883
 

3.创建MqttProperties这个类，将配置文件中的值注入对象中

@Data
@Component
@ConfigurationProperties(prefix = "mqtt")
public class MqttProperties {
/**
  * 账号
  */
 private String username;

 /**
  * 密码
  */
 private String password;
   /**
  * url
  */
 private String[] url;
}

4.创建MqttConfig这个类，进行mqtt相关配置

@Data
@Configuration
public class MqttConfig {
    /**
     * 发布的通道名称
     */
    public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";
    
    /**
     * 订阅消息的通道名称
     */
    public static final String CHANNEL_NAME_IN = "mqttInputChannel";
 
    @Autowired
    private MqttProperties mqttProperties;

/**
  * MQTT连接器选项
  */
 @Bean
 public MqttConnectOptions getMqttConnectOptions() {
  MqttConnectOptions options = new MqttConnectOptions();
  // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
  // 设置为true表示每次连接到服务器都以新的身份连接
  options.setCleanSession(true);
  // 设置连接的用户名
  
  // 设置连接的密码
  options.setUserName(mqttProperties.getUserName());
  options.setPassword(mqttProperties.getpassword().toCharArray());
  options.setServerURIs(mqttProperties.getUrl().toArray(new String[] {}));
  // 设置超时时间 单位为秒
  options.setConnectionTimeout(10);
  options.setMaxInflight(1000);
  options.setAutomaticReconnect(true);
  // 设置会话心跳时间 单位为秒 服务器会每隔20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
  options.setKeepAliveInterval(20);
  return options;
 }

 /**
  * MQTT客户端工厂类，用户生成mqtt客户端
  *
  * @return
  */
 @Bean
 public MqttPahoClientFactory mqttClientFactory() {
  DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
  factory.setConnectionOptions(getMqttConnectOptions());
  return factory;
 }
 
 /**
  * MQTT出站通道
  * @return
  */
 @Bean(name =CHANNEL_NAME_OUT)
 public MessageChannel mqttOutboundChannel() {
  return new DirectChannel();
 }

 /**
  * MQTT出站通道适配器,用于消息推送
  * @return
  */
 @Bean
 @ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
 public MessageHandler mqttOutbound() {
  String clientId = mqttProperties.getClientId().concat("-sender");
  MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId,mqttClientFactory());
  messageHandler.setAsync(true);
  messageHandler.setDefaultQos(1);
  return messageHandler;
 }
 
 
 /**
  * MQTT入站通道
  * @return
  */
    @Bean(name =CHANNEL_NAME_IN)
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

 /**
  * MQTT入站通道适配器, 消息订阅绑定（消费者）
  * @return
  */
 @Bean
 public MessageProducer inbound() {
  //在application.xml中加上监听的topic和clientId
   String[] topicAry = mqttProperties.getTopics().toArray(new String[] {});
   String clientId = mqttProperties.getClientId().concat("-receiver");
  MqttPahoMessageDrivenChannelAdapter adapter = new         MqttPahoMessageDrivenChannelAdapter(
    clientId, mqttClientFactory(),topicAry);
  //设置超时时间，默认30000毫秒
  adapter.setCompletionTimeout(30000L);
  adapter.setConverter(new DefaultPahoMessageConverter());
  //设置接收消息服务质量为1
  adapter.setQos(1);
  adapter.setOutputChannel(mqttInputChannel());
  return adapter;
 }

}

5.创建监听类MqttListener

@Bean
 @ServiceActivator(inputChannel = "mqttInputChannel")
 public MessageHandler handler() {
  return new MessageHandler() {
   
   @Override
   public void handleMessage(Message<?> message) throws MessagingException {
    MessageHeaders headers = message.getHeaders();
    System.err.println(headers);
    System.err.println("接受到消息");
   }
  };
 }

6.创建发送类MqttGateway

@Component
@MessagingGateway(defaultRequestChannel = MqttConfig.CHANNEL_NAME_OUT)
public interface MqttGateway {

 /**
 * 发送信息到MQTT服务器
 *
 * @param topic 主题
 * @param payload 消息主体
 */
 void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);

 /**
 * 发送信息到MQTT服务器
 *
 * @param topic 主题
 * @param qos 对消息处理的几种机制。<br> 0 表示的是订阅者没收到消息不会再次发送，消息会丢失。<br>
 * 1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。<br>
 * 2 多了一次去重的动作，确保订阅者收到的消息有一次。
 * @param payload 消息主体
 */
 void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);

}

在需要发送的业务中调用发送接口，在监听类出进行解析并进行业务处理即可