springboot集成mqtt的实践开发

springboot集成mqtt的实践开发

序

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

maven

org.springframework.boot

spring-boot-starter-integration

org.springframework.integration

spring-integration-stream

org.springframework.integration

spring-integration-mqtt

配置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());

TqRNETqSH 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

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。