一、Spring Integration MQTTSpring Integration提供入站和出站通道适配器以支持消息队列遥测传输MQTT协议。您需要将此依赖项包含到您的项目中dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId version6.5.1/version /dependency当前实现使用Eclipse Paho MQTT Client库。从6.5版开始org. eclipse.pahoorg.eclipse.paho.client.mqttv3依赖项是可选依赖项因此必须明确包含在MQTT v3支持的目标项目中。使用DefaultMqttPahoClientFactory实现两个适配器的配置。有关配置选项的更多信息请参阅Paho留档。1、入站消息驱动通道适配器入站通道适配器由MqttPahoMessageDrivenChannelAdapter实现。为方便起见您可以使用命名空间对其进行配置。最小配置可能如下bean idclientFactory classorg.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory property nameconnectionOptions bean classorg.eclipse.paho.client.mqttv3.MqttConnectOptions property nameuserName value${mqtt.username}/ property namepassword value${mqtt.password}/ /bean /property /bean int-mqtt:message-driven-channel-adapter idmqttInbound client-id${mqtt.default.client.id}.src url${mqtt.url} topicssometopic client-factoryclientFactory channeloutput/以下清单显示了可用属性int-mqtt:message-driven-channel-adapter idoneTopicAdapter client-idfoo urltcp://localhost:1883 topicsbar,baz qos1,2 convertermyConverter client-factoryclientFactory send-timeout123 error-channelerrors recovery-interval10000 manual-acksfalse channelout /从版本4.2.2开始当适配器成功订阅主题时发布MqttSubscribedEvent。连接或订阅失败时发布MqttConnectionFailedEvent事件。这些事件可以由实现Application ationListener的bean接收。此外一个名为恢复间隔的新属性控制适配器在故障后尝试重新连接的间隔。它默认为10000ms十秒钟。2、在运行时添加和删除主题从4.1版开始您可以通过编程方式更改适配器订阅的主题。Spring Integration提供了addTopic和removeTopic方法。添加主题时您可以选择指定QoS默认值1。您还可以通过向具有适当有效负载的Control-bus/发送适当消息来修改主题——例如“myMqttAdapter. addTopicfoo1”。停止和启动适配器对主题列表没有影响它不会恢复到配置中的原始设置。更改不会在应用程序上下文的生命周期之后保留。新的应用程序上下文恢复到配置的设置。在适配器停止或与代理断开连接时更改主题会在下次建立连接时生效。3、手动Acks从5.3版开始您可以将manualAcks属性设置为true。通常用于异步确认传递。当设置为true时会将标头IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK添加到消息中值为SimpleAcknowledgment。您必须调用确认方法来完成传递。有关详细信息请参阅IMqttClient setManualAcks和messageArrivedComplete的Javadocs。为方便起见提供了标头访问器StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge();从版本5.2.11开始当消息转换器抛出异常或从MqttMessage转换返回null时MqttPahoMessageDrivenChannelAdapter将ErrorMessage发送到errorChannel如果提供。否则将重新抛出此转换错误到MQTT客户端回调中。4、使用Java配置SpringBootApplication public class MqttJavaApplication { public static void main(String[] args) { new SpringApplicationBuilder(MqttJavaApplication.class) .web(false) .run(args); } Bean public MessageChannel mqttInputChannel() { return new DirectChannel(); } Bean public MessageProducer inbound() { MqttPahoMessageDrivenChannelAdapter adapter new MqttPahoMessageDrivenChannelAdapter(tcp://localhost:1883, testClient, topic1, topic2); adapter.setCompletionTimeout(5000); adapter.setConverter(new DefaultPahoMessageConverter()); adapter.setQos(1); adapter.setOutputChannel(mqttInputChannel()); return adapter; } Bean ServiceActivator(inputChannel mqttInputChannel) public MessageHandler handler() { return new MessageHandler() { Override public void handleMessage(Message? message) throws MessagingException { System.out.println(message.getPayload()); } }; } }5、使用JavaDSL进行配置以下Spring Boot应用程序提供了使用JavaDSL配置入站适配器的示例SpringBootApplication public class MqttJavaApplication { public static void main(String[] args) { new SpringApplicationBuilder(MqttJavaApplication.class) .web(false) .run(args); } Bean public IntegrationFlow mqttInbound() { return IntegrationFlow.from( new MqttPahoMessageDrivenChannelAdapter(tcp://localhost:1883, testClient, topic1, topic2)) .handle(m - System.out.println(m.getPayload())) .get(); } }6、出站通道适配器出站通道适配器由MqttPahoMessageHandler实现该MqttPahoMessageHandler被包装在消费者端点中。为方便起见您可以使用命名空间对其进行配置。从4.1版开始适配器支持异步发送操作避免在确认交付之前阻塞。如果需要您可以发出应用程序事件以使应用程序能够确认交付。以下清单显示了出站通道适配器可用的属性int-mqtt:outbound-channel-adapter idwithConverter client-idfoo urltcp://localhost:1883 convertermyConverter client-factoryclientFactory default-qos1 qos-expression default-retainedtrue retained-expression default-topicbar topic-expression asyncfalse async-eventsfalse channeltarget /7、使用Java配置以下Spring Boot应用程序显示了如何使用Java配置配置出站适配器的示例SpringBootApplication IntegrationComponentScan public class MqttJavaApplication { public static void main(String[] args) { ConfigurableApplicationContext context new SpringApplicationBuilder(MqttJavaApplication.class) .web(false) .run(args); MyGateway gateway context.getBean(MyGateway.class); gateway.sendToMqtt(foo); } Bean public MqttPahoClientFactory mqttClientFactory() { DefaultMqttPahoClientFactory factory new DefaultMqttPahoClientFactory(); MqttConnectOptions options new MqttConnectOptions(); options.setServerURIs(new String[] { tcp://host1:1883, tcp://host2:1883 }); options.setUserName(username); options.setPassword(password.toCharArray()); factory.setConnectionOptions(options); return factory; } Bean ServiceActivator(inputChannel mqttOutboundChannel) public MessageHandler mqttOutbound() { MqttPahoMessageHandler messageHandler new MqttPahoMessageHandler(testClient, mqttClientFactory()); messageHandler.setAsync(true); messageHandler.setDefaultTopic(testTopic); return messageHandler; } Bean public MessageChannel mqttOutboundChannel() { return new DirectChannel(); } MessagingGateway(defaultRequestChannel mqttOutboundChannel) public interface MyGateway { void sendToMqtt(String data); } }8、使用JavaDSL进行配置以下Spring Boot应用程序提供了使用JavaDSL配置出站适配器的示例SpringBootApplication public class MqttJavaApplication { public static void main(String[] args) { new SpringApplicationBuilder(MqttJavaApplication.class) .web(false) .run(args); } Bean public IntegrationFlow mqttOutboundFlow() { return f - f.handle(new MqttPahoMessageHandler(tcp://host1:1883, someMqttClient)); } }9、Events某些应用程序事件由适配器发布MqttConnectionFailedEvent-如果连接失败或连接随后丢失则由两个适配器发布。对于MQTT v5 Paho客户端当服务器执行正常断开连接时也会发出此事件在这种情况下丢失连接的原因为空。MqttMessageSentEvent-如果在异步模式下运行则在发送消息时由出站适配器发布。MqttMessageDeliveredEvent-当客户端指示消息已传递如果在异步模式下运行时由出站适配器发布。MqttMessageNotDeliveredEvent-当客户端指示消息尚未传递如果在异步模式下运行时由出站适配器发布。MqttSubscribedEvent-由入站适配器在订阅主题后发布。这些事件可以由Application ationListenerMqttIntegrationEvent或EventListener方法接收。要确定事件的来源请使用以下内容您可以检查bean名称和/或连接选项以访问服务器URI等。MqttPahoComponent source event.getSourceAsType(); String beanName source.getBeanName(); MqttConnectOptions options source.getConnectionInfo();9、对MQTT v5 的支持从版本5.5.5开始spring-integration-mqtt模块为MQTT v5协议提供通道适配器实现。org. eclipse.pahoorg.eclipse.paho.mqttv5.client是可选依赖项因此必须明确包含在目标项目中。由于MQTT v5协议支持MQTT消息中的额外任意属性因此引入了MqttHeaderMapper实现以在发布和接收操作时映射到/从标头。默认情况下通过*模式它映射所有接收到的PUBLISH帧属性包括用户属性。在出站端它映射PUBLISH帧的标头子集contentType、mqtt_messageExpiryInterval、mqtt_responseTopic、mqtt_correlationData。MQTT v5协议的出站通道适配器以Mqttv5PahoMessageHandler的形式出现。它需要clientId和MQTT代理URL或MqttConnectionOptions引用。它支持MqttClientPersistence选项可以是异步的在这种情况下可以发出MqttIntegrationEvent对象请参阅asyncEvents选项。如果请求消息有效负载是org. eclipse.paho.mqttv5.Common.MqttMessage则通过内部IMqttAsyncClient按原样发布。如果有效负载是byte[]则按原样使用目标MqttMessage有效负载进行发布。如果有效负载是String则将其转换为byte[]进行发布。其余用例委托给提供的MessageConverter它是一个IntegrationContextUtils。ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME应用程序上下文中的bean。注意当请求的消息有效负载已经是MqttMessage时不使用提供的HeaderMapperMqttProperties。以下JavaDSL配置示例演示了如何在集成流中使用此通道适配器Bean public IntegrationFlow mqttOutFlow() { Mqttv5PahoMessageHandler messageHandler new Mqttv5PahoMessageHandler(MQTT_URL, mqttv5SIout); MqttHeaderMapper mqttHeaderMapper new MqttHeaderMapper(); mqttHeaderMapper.setOutboundHeaderNames(some_user_header, MessageHeaders.CONTENT_TYPE); messageHandler.setHeaderMapper(mqttHeaderMapper); messageHandler.setAsync(true); messageHandler.setAsyncEvents(true); messageHandler.setConverter(mqttStringToBytesConverter()); return f - f.handle(messageHandler); }如果连接在启动时或运行时失败Mqttv5PahoMessageHandler将尝试在产生给该处理程序的下一条消息上重新连接。如果此手动重新连接失败则将连接异常抛回给调用者。在这种情况下将应用标准Spring Integration错误处理过程包括请求处理程序建议例如重试或断路器。在javadocs及其超类Mqttv5PahoMessageHandler中查看更多信息。MQTT v5协议的入站通道适配器以Mqttv5PahoMessageDrivenChannelAdapter的形式存在。它需要clientId和MQTT代理URL或MqttConnectionOptions引用以及订阅和消费的主题。它支持MqttClientPersistence选项默认情况下在内存中。可以配置预期的payloadType默认情况下为byte[]并将其传播到提供的SmartMessageConverter以便从接收到的MqttMessage的byte[]进行转换。如果设置了manualAck选项则IntegrationMessageHeaderAccessor。ACKNOWLEDGMENT_CALLBACK标头被添加到消息中以产生SimpleAcknowledgment的实例。HeaderMapperMqttProperties用于将PUBLISH帧属性包括用户属性映射到目标消息标头中。标准MqttMessage属性例如qos、id、dup、保留以及接收到的主题始终映射到标头。有关更多信息请参阅MqttHeaders。从6.3版开始Mqttv5PahoMessageDrivenChannelAdapter提供了基于MqttSubscription的构造函数用于细粒度配置而不是普通的主题名称。提供这些订阅时无法使用通道适配器的qos选项因为这种qos模式是MqttSubscription API的一部分。以下JavaDSL配置示例演示了如何在集成流程中使用此通道适配器Bean public IntegrationFlow mqttInFlow() { Mqttv5PahoMessageDrivenChannelAdapter messageProducer new Mqttv5PahoMessageDrivenChannelAdapter(MQTT_URL, mqttv5SIin, siTest); messageProducer.setPayloadType(String.class); messageProducer.setMessageConverter(mqttStringToBytesConverter()); messageProducer.setManualAcks(true); return IntegrationFlow.from(messageProducer) .channel(c - c.queue(fromMqttChannel)) .get(); }二、Eclipse Paho Java ClientPahoJava客户端是一个用Java编写的MQTT客户端库用于开发在JVM或其他Java兼容平台如Android上运行的应用程序PahoJava客户端提供了两个APIMqttAsyncClient提供了一个完全异步的API通过注册的回调通知活动的完成。MqttClient是围绕MqttAsyncClient的同步包装器其中函数看起来与应用程序同步。1、使用PahoJava客户端Eclipse为那些想要使用Maven来管理其依赖项的人托管一个Nexus存储库。Maven Central存储库中也提供了已发布的库。将存储库定义和下面显示的依赖项定义添加到您的pom. xml。将%REPOURL%替换为https://repo.eclipse.org/content/repositories/paho-releases/用于正式版本或https://repo.eclipse.org/content/repositories/paho-snapshots/用于夜间快照。将%VERSION%替换为所需级别。最新版本1.2.0当前快照版本1.2.1。project ... repositories repository idEclipse Paho Repo/id url%REPOURL%/url /repository /repositories ... dependencies dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version%VERSION%/version /dependency /dependencies /project** MQTTv5客户端的依赖定义 **project ... repositories repository idEclipse Paho Repo/id url%REPOURL%/url /repository /repositories ... dependencies dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.mqttv5.client/artifactId version%VERSION%/version /dependency /dependencies /project2、开发实例下面包含的代码是一个非常基本的示例它连接到服务器并使用MqttClient同步API发布消息。可以在源代码的org. eclipse.paho.sample.mqttv3app目录中找到演示异步API使用的更广泛的示例。import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; public class MqttPublishSample { public static void main(String[] args) { String topic MQTT Examples; String content Message from MqttPublishSample; int qos 2; String broker tcp://mqtt.eclipseprojects.io:1883; String clientId JavaSample; MemoryPersistence persistence new MemoryPersistence(); try { MqttClient sampleClient new MqttClient(broker, clientId, persistence); MqttConnectOptions connOpts new MqttConnectOptions(); connOpts.setCleanSession(true); System.out.println(Connecting to broker: broker); sampleClient.connect(connOpts); System.out.println(Connected); System.out.println(Publishing message: content); MqttMessage message new MqttMessage(content.getBytes()); message.setQos(qos); sampleClient.publish(topic, message); System.out.println(Message published); sampleClient.disconnect(); System.out.println(Disconnected); System.exit(0); } catch(MqttException me) { System.out.println(reason me.getReasonCode()); System.out.println(msg me.getMessage()); System.out.println(loc me.getLocalizedMessage()); System.out.println(cause me.getCause()); System.out.println(excep me); me.printStackTrace(); } } }