用Java的springboot写个mqtt的同时订阅的例子

以下是使用Java Spring Boot编写MQTT同时订阅的示例代码：

首先，需要添加以下依赖项：

<dependency>
   <groupId>org.springframework.integration</groupId>
   <artifactId>spring-integration-mqtt</artifactId>
   <version>5.1.0.RELEASE</version>
</dependency>

然后，在Spring Boot应用程序中创建一个订阅者类，以处理来自MQTT服务器的消息：

import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;

@MessageEndpoint
public class MqttSubscriber {

   @ServiceActivator(inputChannel = "mqttInputChannel")
   public void receiveMessage(String message) {
      System.out.println("Received message: " + message);
   }
}

这里使用了Spring Integration的注释来定义一个消息端点和一个服务活动器。服务活动器接收来自MQTT服务器的消息并将其打印到控制台。

接下来，需要配置MQTT连接和订阅信息：

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.messaging.MessageChannel;

@Configuration
@IntegrationComponentScan
public class MqttConfig {

   @Value("${mqtt.broker.url}")
   private String brokerUrl;

   @Value("${mqtt.username}")
   private String username;

   @Value("${mqtt.password}")
   private String password;

   @Value("${mqtt.topic}")
   private String topic;

   @Value("${mqtt.client.id}")
   private String clientId;

   @Bean
   public MqttPahoClientFactory mqttClientFactory() {
      DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
      factory.setServerURIs(brokerUrl);
      factory.setUserName(username);
      factory.setPassword(password);
      return factory;
   }

   @Bean
   public MessageChannel mqttInputChannel() {
      return new DirectChannel();
   }

   @Bean
   public MessageProducer inbound() {
      MqttPahoMessageDrivenChannelAdapter adapter =
            new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClientFactory(), topic);
      adapter.setCompletionTimeout(5000);
      adapter.setConverter(new DefaultPahoMessageConverter());
      adapter.setQos(1);
      adapter.setOutputChannel(mqttInputChannel());
      return adapter;
   }
}

这里定义了一个MQTT配置类，其中封装了一些连接参数和订阅信息。使用@Value注释将这些值注入到Spring Boot应用程序的属性文件中。

在配置类中，创建了一个DefaultMqttPahoClientFactory来构建MQTT客户端，使用了MqttPahoMessageDrivenChannelAdapter来订阅指定的主题并将消息转发到mqttInputChannel通道。接下来，需要在Spring Boot应用程序的主类中启动消息订阅：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

@SpringBootApplication
public class MqttSubscriberApplication {

   @Autowired
   private DirectChannel mqttInputChannel;

   public static void main(String[] args) {
      ConfigurableApplicationContext context = SpringApplication.run(MqttSubscriberApplication.class, args);
      MqttSubscriberApplication application = context.getBean(MqttSubscriberApplication.class);
      application.sendMessage("Hello MQTT");
   }

   public void sendMessage(String message) {
      Message<String> mqttMessage = MessageBuilder.withPayload(message).build();
      mqttInputChannel.send(mqttMessage);
   }
}

这里使用Spring Boot的@Autowired注释将mqttInputChannel注入到应用程序中，并调用sendMessage方法向MQTT服务器发送消息。一旦消息被发送，MqttSubscriber类中的receiveMessage方法会被触发，将接收到的消息打印到控制台。

以上就是使用Java Spring Boot编写MQTT同时订阅的示例代码。