MQTT 支持
Spring Integration 提供了入站和出站通道适配器来支持消息队列遥测传输 (MQTT) 协议。
您需要将此依赖项包含到您的项目中
-
Maven
-
Gradle
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
<version>6.4.4</version>
</dependency>
compile "org.springframework.integration:spring-integration-mqtt:6.4.4"
当前实现使用了 Eclipse Paho MQTT 客户端 库。
XML 配置和本章的大部分内容都关于 MQTT v3.1 协议支持和相应的 Paho 客户端。有关相应的协议支持,请参阅 MQTT v5 支持 段落。 |
两个适配器的配置都是通过使用 DefaultMqttPahoClientFactory
来实现的。有关配置选项的更多信息,请参阅 Paho 文档。
我们建议配置一个 MqttConnectOptions 对象并将其注入到工厂中,而不是在工厂本身上设置(已弃用的)选项。 |
入站(消息驱动)通道适配器
入站通道适配器由 MqttPahoMessageDrivenChannelAdapter
实现。为了方便起见,您可以使用命名空间对其进行配置。一个最小配置可能如下所示
<bean id="clientFactory"
class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="connectionOptions">
<bean class="org.eclipse.paho.client.mqttv3.MqttConnectOptions">
<property name="userName" value="${mqtt.username}"/>
<property name="password" value="${mqtt.password}"/>
</bean>
</property>
</bean>
<int-mqtt:message-driven-channel-adapter id="mqttInbound"
client-id="${mqtt.default.client.id}.src"
url="${mqtt.url}"
topics="sometopic"
client-factory="clientFactory"
channel="output"/>
以下列表显示了可用属性
<int-mqtt:message-driven-channel-adapter id="oneTopicAdapter"
client-id="foo" (1)
url="tcp://localhost:1883" (2)
topics="bar,baz" (3)
qos="1,2" (4)
converter="myConverter" (5)
client-factory="clientFactory" (6)
send-timeout="123" (7)
error-channel="errors" (8)
recovery-interval="10000" (9)
manual-acks="false" (10)
channel="out" />
1 | 客户端 ID。 |
2 | 代理 URL。 |
3 | 此适配器接收消息的主题列表,以逗号分隔。 |
4 | QoS 值列表,以逗号分隔。它可以是应用于所有主题的单个值,也可以是每个主题的值(在这种情况下,列表长度必须相同)。 |
5 | 一个可选的 MqttMessageConverter 。默认情况下,默认的 DefaultPahoMessageConverter 生成的消息其 payload 为 String ,并带有以下消息头:
|
6 | 客户端工厂。 |
7 | send() 超时时间。仅当通道可能阻塞时(例如当前已满的有界 QueueChannel )才适用。 |
8 | 错误通道。如果提供了错误通道,下游异常将以 ErrorMessage 的形式发送到此通道。Payload 是一个 MessagingException ,其中包含失败的消息和原因。 |
9 | 恢复间隔。它控制适配器在失败后尝试重新连接的间隔。默认为 10000ms (十秒)。 |
10 | 确认模式;设置为 true 表示手动确认。 |
从 4.1 版本开始,您可以省略 URL。相反,您可以在 DefaultMqttPahoClientFactory 的 serverURIs 属性中提供服务器 URI。例如,这样做可以连接到高可用 (HA) 集群。 |
从 4.2.2 版本开始,当适配器成功订阅主题时,会发布 MqttSubscribedEvent
。当连接或订阅失败时,会发布 MqttConnectionFailedEvent
事件。实现 ApplicationListener
的 bean 可以接收这些事件。
此外,一个新的属性 recoveryInterval
控制适配器在失败后尝试重新连接的间隔。默认为 10000ms
(十秒)。
在 4.2.3 版本之前,适配器停止时客户端总是取消订阅。这是不正确的,因为如果客户端的 QOS 大于 0,我们需要保持订阅活动状态,以便适配器停止时到达的消息在下次启动时能够送达。这还需要将客户端工厂的 从 4.2.3 版本开始,如果 可以通过设置工厂的 要恢复到 4.2.3 版本之前的行为,请使用 |
从 5.0 版本开始, |
运行时添加和移除主题
从 4.1 版本开始,您可以通过编程方式更改适配器订阅的主题。Spring Integration 提供了 addTopic()
和 removeTopic()
方法。添加主题时,您可以选择指定 QoS
(默认值:1)。您还可以通过向 <control-bus/>
发送适当的消息以及适当的 payload 来修改主题——例如:"myMqttAdapter.addTopic('foo', 1)"
。
停止和启动适配器对主题列表没有影响(它不会恢复到配置中的原始设置)。这些更改不会在应用程序上下文的生命周期结束后保留。新的应用程序上下文会恢复到配置的设置。
在适配器停止(或与代理断开连接)期间更改主题,将在下次建立连接时生效。
手动确认 (Acks)
从 5.3 版本开始,您可以将 manualAcks
属性设置为 true。这通常用于异步确认送达。设置为 true
时,消息头(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK
)会添加到消息中,其值为一个 SimpleAcknowledgment
。您必须调用 acknowledge()
方法来完成送达。有关更多信息,请参阅 IMqttClient
的 setManualAcks()
和 messageArrivedComplete()
的 Javadocs。为方便起见,提供了消息头访问器。
StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge();
从 5.2.11
版本开始,当消息转换器抛出异常或从 MqttMessage
转换返回 null
时,MqttPahoMessageDrivenChannelAdapter
(如果提供了错误通道)会将 ErrorMessage
发送到 errorChannel
。否则,它会将此转换错误重新抛出到 MQTT 客户端回调中。
使用 Java 配置进行配置
以下 Spring Boot 应用程序展示了如何使用 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());
}
};
}
}
使用 Java DSL 进行配置
以下 Spring Boot 应用程序提供了使用 Java DSL 配置入站适配器的示例:
@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();
}
}
出站通道适配器
出站通道适配器由 MqttPahoMessageHandler
实现,它被包装在一个 ConsumerEndpoint
中。为了方便起见,您可以使用命名空间进行配置。
从 4.1 版本开始,适配器支持异步发送操作,避免阻塞直到确认送达。您可以发出应用程序事件,以便应用程序在需要时确认送达。
以下列表显示了出站通道适配器的可用属性:
<int-mqtt:outbound-channel-adapter id="withConverter"
client-id="foo" (1)
url="tcp://localhost:1883" (2)
converter="myConverter" (3)
client-factory="clientFactory" (4)
default-qos="1" (5)
qos-expression="" (6)
default-retained="true" (7)
retained-expression="" (8)
default-topic="bar" (9)
topic-expression="" (10)
async="false" (11)
async-events="false" (12)
channel="target" />
1 | 客户端 ID。 |
2 | 代理 URL。 |
3 | 一个可选的 MqttMessageConverter 。默认的 DefaultPahoMessageConverter 识别以下消息头:
|
4 | 客户端工厂。 |
5 | 默认的服务质量。如果在 mqtt_qos 消息头中未找到,或 qos-expression 返回 null 时使用。如果您提供了自定义的 converter 则不使用。 |
6 | 用于评估确定 qos 的表达式。默认值为 headers[mqtt_qos] 。 |
7 | retained 标志的默认值。如果在 mqtt_retained 消息头中未找到时使用。如果您提供了自定义的 converter 则不使用。 |
8 | 用于评估确定 retained 布尔值的表达式。默认值为 headers[mqtt_retained] 。 |
9 | 消息发送的默认主题(如果在 mqtt_topic 消息头中未找到时使用)。 |
10 | 用于评估确定目标主题的表达式。默认值为 headers['mqtt_topic'] 。 |
11 | 当设置为 true 时,调用者不会阻塞。相反,它会在发送消息时等待送达确认。默认值为 false (发送会阻塞直到确认送达)。 |
12 | 当 async 和 async-events 都为 true 时,会发出 MqttMessageSentEvent (参见 事件)。它包含消息、主题、客户端库生成的 messageId 、clientId 以及 clientInstance (每次客户端连接时递增)。当客户端库确认送达时,会发出 MqttMessageDeliveredEvent 。它包含 messageId 、clientId 和 clientInstance ,以便将送达与 send() 关联起来。任何 ApplicationListener 或事件入站通道适配器都可以接收这些事件。请注意,MqttMessageDeliveredEvent 可能在 MqttMessageSentEvent 之前收到。默认值为 false 。 |
从 4.1 版本开始,可以省略 URL。相反,可以在 DefaultMqttPahoClientFactory 的 serverURIs 属性中提供服务器 URI。例如,这使得连接到高可用 (HA) 集群成为可能。 |
使用 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);
}
}
使用 Java DSL 进行配置
以下 Spring Boot 应用程序提供了使用 Java DSL 配置出站适配器的示例:
@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"));
}
}
事件
适配器会发布某些应用程序事件。
-
MqttConnectionFailedEvent
- 如果连接失败或随后连接断开,由两个适配器发布。对于 MQTT v5 Paho 客户端,当服务器执行正常断开连接时也会发出此事件,此时连接断开的cause
为null
。 -
MqttMessageSentEvent
- 在异步模式下运行时,当消息已发送时由出站适配器发布。 -
MqttMessageDeliveredEvent
- 在异步模式下运行时,当客户端指示消息已送达时由出站适配器发布。 -
MqttMessageNotDeliveredEvent
- 在异步模式下运行时,当客户端指示消息未送达时由出站适配器发布。 -
MqttSubscribedEvent
- 入站适配器订阅主题后发布。
这些事件可以由 ApplicationListener<MqttIntegrationEvent>
或使用 `@EventListener` 方法接收。
要确定事件的来源,请使用以下方法;您可以检查 bean 名称和/或连接选项(以访问服务器 URI 等)。
MqttPahoComponent source = event.getSourceAsType();
String beanName = source.getBeanName();
MqttConnectOptions options = source.getConnectionInfo();
MQTT v5 支持
从 5.5.5 版本开始,spring-integration-mqtt
模块提供了针对 MQTT v5 协议的通道适配器实现。org.eclipse.paho:org.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
选项,可以是 async
的,并且在这种情况下可以发出 MqttIntegrationEvent
对象(参见 asyncEvents
选项)。如果请求消息的 payload 是一个 org.eclipse.paho.mqttv5.common.MqttMessage
,它将通过内部的 IMqttAsyncClient
原样发布。如果 payload 是 byte[]
,它将原样用于目标 MqttMessage
的 payload 进行发布。如果 payload 是 String
,它将被转换为 byte[]
进行发布。其余的使用场景被委托给提供的 MessageConverter
,它是来自应用程序上下文的 IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME
ConfigurableCompositeMessageConverter
bean。注意:当请求的消息 payload 已经是 MqttMessage
时,提供的 HeaderMapper<MqttProperties>
不会被使用。以下 Java DSL 配置示例演示了如何在集成流中使用此通道适配器:
@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);
}
org.springframework.integration.mqtt.support.MqttMessageConverter 不能与 Mqttv5PahoMessageHandler 一起使用,因为它的契约仅针对 MQTT v3 协议。 |
如果连接在启动时或运行时失败,Mqttv5PahoMessageHandler
会在发送到此 handler 的下一条消息时尝试重新连接。如果此手动重新连接失败,则连接异常会被抛回给调用者。在这种情况下,会应用标准的 Spring Integration 错误处理过程,包括请求 handler advices,例如 retry 或 circuit breaker。
有关更多信息,请参阅 Mqttv5PahoMessageHandler
的 javadocs 及其父类。
MQTT v5 协议的入站通道适配器是 Mqttv5PahoMessageDrivenChannelAdapter
。它需要 clientId
和 MQTT 代理 URL 或 MqttConnectionOptions
引用,以及要订阅和消费的主题。它支持 MqttClientPersistence
选项,该选项默认是内存中的。可以配置预期的 payloadType
(默认为 byte[]
),它会被传递给提供的 SmartMessageConverter
,用于从收到的 MqttMessage
的 byte[]
进行转换。如果设置了 manualAck
选项,则会在消息中添加 IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK
消息头,其值是 SimpleAcknowledgment
的一个实例。HeaderMapper<MqttProperties>
用于将 PUBLISH
帧属性(包括用户属性)映射到目标消息头中。标准的 MqttMessage
属性,如 qos
、id
、dup
、retained
,以及收到的主题,总是被映射到消息头中。有关更多信息,请参见 MqttHeaders
。
从 6.3 版本开始,Mqttv5PahoMessageDrivenChannelAdapter
提供了基于 MqttSubscription
的构造函数,用于细粒度配置,而不是简单的通用主题名称。当提供了这些订阅时,通道适配器的 qos
选项将无法使用,因为这种 qos
模式是 MqttSubscription
API 的一部分。
以下 Java DSL 配置示例演示了如何在集成流中使用此通道适配器:
@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();
}
org.springframework.integration.mqtt.support.MqttMessageConverter 不能与 Mqttv5PahoMessageDrivenChannelAdapter 一起使用,因为它的契约仅针对 MQTT v3 协议。 |
有关更多信息,请参阅 Mqttv5PahoMessageDrivenChannelAdapter
的 javadocs 及其父类。
建议将 MqttConnectionOptions#setAutomaticReconnect(boolean) 设置为 true,以让内部的 IMqttAsyncClient 实例处理重新连接。否则,只有手动重新启动 Mqttv5PahoMessageDrivenChannelAdapter 才能处理重新连接,例如通过 MqttConnectionFailedEvent 处理断开连接。 |
共享 MQTT 客户端支持
如果多个集成需要使用单个 MQTT ClientID,则不能使用多个 MQTT 客户端实例,因为 MQTT 代理可能对每个 ClientID 的连接数量有限制(通常只允许一个连接)。为了让单个客户端被不同的通道适配器重用,可以使用 org.springframework.integration.mqtt.core.ClientManager
组件并将其传递给所需的任何通道适配器。它将管理 MQTT 连接生命周期并在需要时进行自动重新连接。此外,还可以为客户端管理器提供自定义连接选项和 MqttClientPersistence
,就像当前可以为通道适配器组件所做的那样。
注意,同时支持 MQTT v5 和 v3 通道适配器。
以下 Java DSL 配置示例演示了如何在集成流中使用此客户端管理器:
@Bean
public ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager() {
MqttConnectionOptions connectionOptions = new MqttConnectionOptions();
connectionOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
connectionOptions.setConnectionTimeout(30000);
connectionOptions.setMaxReconnectDelay(1000);
connectionOptions.setAutomaticReconnect(true);
Mqttv5ClientManager clientManager = new Mqttv5ClientManager(connectionOptions, "client-manager-client-id-v5");
clientManager.setPersistence(new MqttDefaultFilePersistence());
return clientManager;
}
@Bean
public IntegrationFlow mqttInFlowTopic1(
ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager) {
Mqttv5PahoMessageDrivenChannelAdapter messageProducer =
new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, "topic1");
return IntegrationFlow.from(messageProducer)
.channel(c -> c.queue("fromMqttChannel"))
.get();
}
@Bean
public IntegrationFlow mqttInFlowTopic2(
ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager) {
Mqttv5PahoMessageDrivenChannelAdapter messageProducer =
new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, "topic2");
return IntegrationFlow.from(messageProducer)
.channel(c -> c.queue("fromMqttChannel"))
.get();
}
@Bean
public IntegrationFlow mqttOutFlow(
ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager) {
return f -> f.handle(new Mqttv5PahoMessageHandler(clientManager));
}
从 6.4 版本开始,现在可以通过对应的 ClientManager 和 IntegrationFlowContext 在运行时添加 MqttPahoMessageDrivenChannelAdapter 和 Mqttv5PahoMessageDrivenChannelAdapter 的多个实例。 |
private void addAddRuntimeAdapter(IntegrationFlowContext flowContext, Mqttv5ClientManager clientManager,
String topic, MessageChannel channel) {
flowContext
.registration(
IntegrationFlow
.from(new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, topic))
.channel(channel)
.get())
.register();
}