springboot集成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>

  @Bean  public MqttPahoClientFactory mqttClientFactory() {    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();    factory.setServerURIs("tcp://demo:1883");//    factory.setUserName("guest");//    factory.setPassword("guest");    return factory;  }

  @Bean  public IntegrationFlow mqttInFlow() {    return IntegrationFlows.from(mqttInbound())        .transform(p -> p + ", received from MQTT")        .handle(logger())        .get();  }  private LoggingHandler logger() {    LoggingHandler loggingHandler = new LoggingHandler("INFO");    loggingHandler.setLoggerName("siSample");    return loggingHandler;  }  @Bean  public MessageProducerSupport mqttInbound() {    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("siSampleConsumer",        mqttClientFactory(), "siSampleTopic");    adapter.setCompletionTimeout(5000);    adapter.setConverter(new DefaultPahoMessageConverter());    adapter.setQos(1);    return adapter;  }

@Bean  public IntegrationFlow mqttOutFlow() {    //console input//    return IntegrationFlows.from(CharacterStreamReadingMessageSource.stdin(),//        e -> e.poller(Pollers.fixedDelay(1000)))//        .transform(p -> p + " sent to MQTT")//        .handle(mqttOutbound())//        .get();    return IntegrationFlows.from(outChannel())        .handle(mqttOutbound())        .get();  }    @Bean  public MessageChannel outChannel() {    return new DirectChannel();  }  @Bean  public MessageHandler mqttOutbound() {    MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("siSamplePublisher", mqttClientFactory());    messageHandler.setAsync(true);    messageHandler.setDefaultTopic("siSampleTopic");    return messageHandler;  }

@MessagingGateway(defaultRequestChannel = "outChannel")public interface MsgWriter {  void write(String note);}