|
此版本仍在开发中，尚不被认为是稳定的。对于最新的稳定版本，请使用 Spring Integration 6.5.1！ |
MQTT 支持
Spring Integration 提供入站和出站通道适配器来支持消息队列遥测传输 （MQTT） 协议。
您需要将此依赖项包含在您的项目中：
-
Maven
-
Gradle
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
<version>6.5.2-SNAPSHOT</version>
</dependency>compile "org.springframework.integration:spring-integration-mqtt:6.5.2-SNAPSHOT"当前实现使用 Eclipse Paho MQTT 客户端库。
从 6.5 版开始，org.eclipse.paho:org.eclipse.paho.client.mqttv3依赖关系是一个optional依赖项，因此必须显式包含在目标项目中才能支持 MQTT v3。
| 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生成一条消息，其中包含Stringpayload 的标头：
|
| 6 | 客户端工厂。 |
| 7 | 这send()超时。
仅当通道可能阻塞（例如有界的QueueChannel目前已满）。 |
| 8 | 错误通道。
下游异常（如果提供）将发送到此通道，则在ErrorMessage.
有效负载是MessagingException，其中包含失败的消息和原因。 |
| 9 | 恢复间隔。
它控制适配器在发生故障后尝试重新连接的时间间隔。
它默认为10000ms（十秒）。 |
| 10 | 确认方式;设置为 true 以进行手动确认。 |
从 4.1 版开始，您可以省略 URL。
相反，您可以在serverURIs属性的DefaultMqttPahoClientFactory.
例如，这样做可以启用与高可用性 （HA） 集群的连接。 |
从 4.2.2 版开始，MqttSubscribedEvent当适配器成功订阅主题时，将发布。MqttConnectionFailedEvent当连接或订阅失败时，将发布事件。
这些事件可以由实现ApplicationListener.
此外，一个名为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/>具有适当的有效负载 — 例如："myMqttAdapter.addTopic('foo', 1)".
停止和启动适配器对主题列表没有影响（它不会恢复到配置中的原始设置）。更改不会在应用程序上下文的生命周期之后保留。新的应用程序上下文将恢复为配置的设置。
在适配器停止（或与代理断开连接）时更改主题将在下次建立连接时生效。
手动 Acks
从 5.3 版开始，您可以将manualAcks属性设置为 true。通常用于异步确认传递。当设置为true、标题 （IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK） 被添加到消息中，值为SimpleAcknowledgment. 您必须调用acknowledge()方法来完成交付。请参阅 JavadocsIMqttClient setManualAcks()和messageArrivedComplete()了解更多信息。为方便起见，提供了标头访问器：
StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge();从版本开始5.2.11，当消息转换器抛出异常或返回null从MqttMessage转换，则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 | 保留标志的默认值。如果没有，则使用它mqtt_retainedheader。如果自定义converter被提供。 |
| 8 | 要计算以确定保留的布尔值的表达式。默认值为headers[mqtt_retained]. |
| 9 | 将消息发送到的默认主题（如果没有mqtt_topicheader 中找到）。 |
| 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。
相反，可以在serverURIs属性的DefaultMqttPahoClientFactory.
例如，这支持与高可用性 （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是一个optional依赖项，因此必须显式包含在目标项目中。
由于 MQTT v5 协议支持 MQTT 消息中的额外任意属性，因此MqttHeaderMapper引入了在发布和接收作上映射到/映射标头的实现。
默认情况下，（通过模式）它映射所有接收到*PUBLISH框架属性（包括用户属性）。
在出站端，它映射了PUBLISH框架：contentType,mqtt_messageExpiryInterval,mqtt_responseTopic,mqtt_correlationData.
MQTT v5 协议的出站通道适配器以Mqttv5PahoMessageHandler.
它需要一个clientId和 MQTT 代理 URL 或MqttConnectionOptions参考。
它支持MqttClientPersistence选项，可以是async并能发射MqttIntegrationEvent在这种情况下，对象（请参阅asyncEvents选项）。
如果请求消息有效负载是org.eclipse.paho.mqttv5.common.MqttMessage，它通过内部按原样发布IMqttAsyncClient.
如果有效负载是byte[]它按原样用于目标MqttMessage有效负载。
如果有效负载是String它被转换为byte[]发布。
其余用例被委托给提供的MessageConverter这是一个IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME ConfigurableCompositeMessageConverterbean 来自应用程序上下文。
注意：提供的HeaderMapper<MqttProperties>当请求的消息有效负载已经是MqttMessage.
以下 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尝试在生成到此处理程序的下一条消息上重新连接。
如果此手动重新连接失败，则连接异常将抛回给调用方。
在这种情况下，应用了标准的 Spring Integration 错误处理过程，包括请求处理程序通知，例如重试或断路器。
在Mqttv5PahoMessageHandlerjavadocs 及其超类。
MQTT v5 协议的入站通道适配器以Mqttv5PahoMessageDrivenChannelAdapter.
它需要一个clientId和 MQTT 代理 URL 或MqttConnectionOptions引用，以及要订阅和使用的主题。
它支持MqttClientPersistence选项，默认情况下在内存中。
预期的payloadType (byte[]默认情况下）可以配置，并且它会传播到提供的SmartMessageConverter用于转换byte[]的MqttMessage.
如果manualAck选项，则IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACKheader 添加到消息中以生成为SimpleAcknowledgment.
这HeaderMapper<MqttProperties>用于映射PUBLISH框架属性（包括用户属性）到目标邮件头中。
标准MqttMessage属性，例如qos,id,dup,retained，加上收到的主题始终映射到标头。
看MqttHeaders了解更多信息。
从 6.3 版开始，Mqttv5PahoMessageDrivenChannelAdapter提供基于MqttSubscription用于细粒度配置，而不是纯主题名称。
提供这些订阅后，qos不能使用通道适配器的选项，因为这样的qos模式是MqttSubscription应用程序接口。
以下 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 协议。 |
在Mqttv5PahoMessageDrivenChannelAdapterjavadocs 及其超类。
建议拥有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 版开始，多个实例MqttPahoMessageDrivenChannelAdapter和Mqttv5PahoMessageDrivenChannelAdapter现在可以在运行时使用相应的ClientManager通过IntegrationFlowContext
|
private void addAddRuntimeAdapter(IntegrationFlowContext flowContext, Mqttv5ClientManager clientManager,
String topic, MessageChannel channel) {
flowContext
.registration(
IntegrationFlow
.from(new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, topic))
.channel(channel)
.get())
.register();
}