springboot集成mqtt的实践开发

序

MQTT（Message Queuing Telemetry Transport）是基于二进制消息的发布/订阅编程模式的消息协议，非常适合需要低功耗和网络带宽有限的IoT场景。这里简单介绍一下如何在springboot中集成。

maven

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

配置client factory

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

配置consumer

  @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;
  }

配置producer

@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

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

这样就大功告成了

doc

spring-integration-mqtt

spring-integration-samples-mqtt

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。

# springboot集成mqtt # SpringBoot中集成MQTT # SpringBoot整合MQTT并实现异步线程调用的问题 # SpringBoot集成mqtt的多模块项目配置详解 # springboot 实现mqtt物联网的示例代码 # SpringBoot+MQTT+apollo实现订阅发布功能的示例 # SpringBoot+netty-socketio实现服务器端消息推送 # SpringBoot+WebSocket+Netty实现消息推送的示例代码 # SpringBoot实现钉钉机器人消息推送的示例代码 # springboot整合mqtt实现消息订阅和推送功能 # 大功告成 # 介绍一下 # 网络带宽 # 大家多多 # 低功耗 # 如何在 # 是基于 # integration # stream # MqttPahoClientFactory # DefaultMqttPahoClientFactory # mqttClientFactory # public # Bean # java # factory # mqtt # client # setServerURIs # consumer

返回目录在线咨询

上一篇：如何将SEO关键词应用于电子邮件营销_SEO优化教程
下一篇：网站服务器迁移需遵循哪些关键步骤避免数据丢失？

您的项目需求

*请认真填写需求信息，我们会在24小时内与您取得联系。

全网整合营销服务商

springboot集成mqtt的实践开发

您的项目需求