MQTT Support

Spring Integration 提供入站和出站通道适配器来支持消息队列遥测传输 (MQTT) 协议。

Spring Integration provides inbound and outbound channel adapters to support the Message Queueing Telemetry Transport (MQTT) protocol.

你需要将此依赖项包含在你的项目中：

You need to include this dependency into your project:

Maven
Gradle

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
    <version>{project-version}</version>
</dependency>

compile "org.springframework.integration:spring-integration-mqtt:{project-version}"

当前实现使用 Eclipse Paho MQTT Client库。

The current implementation uses the Eclipse Paho MQTT Client library.

XML 配置和本章的大部分内容都与 MQTT v3.1 协议支持和各自的 Paho Client 有关。有关各自的协议支持，请参阅 MQTT v5 Support 段落。

The XML configuration and most of this chapter are about MQTT v3.1 protocol support and respective Paho Client. See MQTT v5 Support paragraph for respective protocol support.

这两个适配器的配置都使用 DefaultMqttPahoClientFactory 来实现。有关配置选项的详细信息，请参阅 Paho 文档。

Configuration of both adapters is achieved using the DefaultMqttPahoClientFactory. Refer to the Paho documentation for more information about configuration options.

我们建议配置 MqttConnectOptions 对象并将其注入到工厂，而不是设置工厂本身上的（已弃用）选项。

We recommend configuring an MqttConnectOptions object and injecting it into the factory, instead of setting the (deprecated) options on the factory itself.

Inbound (Message-driven) Channel Adapter

入站通道适配器由 MqttPahoMessageDrivenChannelAdapter 实现。为方便起见，您可以使用以下名称空间来配置它。最基本的配置如下所示：

The inbound channel adapter is implemented by the MqttPahoMessageDrivenChannelAdapter. For convenience, you can configure it by using the namespace. A minimal configuration might be as follows:

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

以下清单显示了可用的属性：

The following listing shows the available attributes:

<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	The client ID.
2	The broker URL.
3	A comma-separated list of topics from which this adapter receives messages.
4	A comma-separated list of QoS values. It can be a single value that is applied to all topics or a value for each topic (in which case, the lists must be the same length).
5	An `MqttMessageConverter` (optional). By default, the default `DefaultPahoMessageConverter` produces a message with a `String` payload with the following headers: `mqtt_topic`: The topic from which the message was received `mqtt_duplicate`: `true` if the message is a duplicate `mqtt_qos`: The quality of service You can configure the `DefaultPahoMessageConverter` to return the raw `byte[]` in the payload by declaring it as a `<bean/>` and setting the `payloadAsBytes` property to `true`.
6	The client factory.
7	The `send()` timeout. It applies only if the channel might block (such as a bounded `QueueChannel` that is currently full).
8	The error channel. Downstream exceptions are sent to this channel, if supplied, in an `ErrorMessage`. The payload is a `MessagingException` that contains the failed message and cause.
9	The recovery interval. It controls the interval at which the adapter attempts to reconnect after a failure. It defaults to `10000ms` (ten seconds).
10	The acknowledgment mode; set to true for manual acknowledgment.

从版本 4.1 开始，你可以省略 URL。相反，你可以在 DefaultMqttPahoClientFactory 的 serverURIs 属性中提供服务器 URI。这样做可以实现连接到高可用性 (HA) 群集。

Starting with version 4.1, you can omit the URL. Instead, you can provide the server URIs in the serverURIs property of the DefaultMqttPahoClientFactory. Doing so enables, for example, connection to a highly available (HA) cluster.

从 4.2.2 版开始，当适配器成功订阅主题后，将发布 MqttSubscribedEvent。当连接或订阅失败时，将发布 MqttConnectionFailedEvent 事件。这些事件可以通过实现 ApplicationListener 的 bean 来接收。

Starting with version 4.2.2, an MqttSubscribedEvent is published when the adapter successfully subscribes to the topics. MqttConnectionFailedEvent events are published when the connection or subscription fails. These events can be received by a bean that implements ApplicationListener.

此外，一个名为 recoveryInterval 的新属性控制了适配器在出现故障后尝试重新连接的时间间隔。它的默认值为 10000ms（十秒）。

Also, a new property called recoveryInterval controls the interval at which the adapter attempts to reconnect after a failure. It defaults to 10000ms (ten seconds).

在 4.2.3 版之前，当适配器停止时，客户端始终取消订阅。这是不正确的，因为如果客户端 QOS 大于 0，我们就需要保持订阅活动，这样在适配器停止时到达的消息将在下次启动时被传递。这也要求在客户端工厂上将 cleanSession 属性设置为 false。它默认为 true。

Prior to version 4.2.3, the client always unsubscribed when the adapter was stopped. This was incorrect because, if the client QOS is greater than 0, we need to keep the subscription active so that messages arriving while the adapter is stopped are delivered on the next start. This also requires setting the cleanSession property on the client factory to false. It defaults to true.

从 4.2.3 版开始，如果 cleanSession 属性为 false，适配器不会（默认）取消订阅。

Starting with version 4.2.3, the adapter does not unsubscribe (by default) if the cleanSession property is false.

可以通过在工厂上设置 consumerCloseAction 属性来覆盖此行为。它可以有以下值：UNSUBSCRIBE_ALWAYS、UNSUBSCRIBE_NEVER 和 UNSUBSCRIBE_CLEAN。后者（默认值）仅在 cleanSession 属性为 true 时取消订阅。

This behavior can be overridden by setting the consumerCloseAction property on the factory. It can have values: UNSUBSCRIBE_ALWAYS, UNSUBSCRIBE_NEVER, and UNSUBSCRIBE_CLEAN. The latter (the default) unsubscribes only if the cleanSession property is true.

要恢复到 4.2.3 之前的行为，请使用 UNSUBSCRIBE_ALWAYS。

To revert to the pre-4.2.3 behavior, use UNSUBSCRIBE_ALWAYS.

从版本 5.0 开始，topic、qos 和 retained 属性映射到 .RECEIVED_… 标头（MqttHeaders.RECEIVED_TOPIC、MqttHeaders.RECEIVED_QOS 和 MqttHeaders.RECEIVED_RETAINED），以避免无意中传播到（默认情况下）使用 MqttHeaders.TOPIC、MqttHeaders.QOS 和 MqttHeaders.RETAINED 标头的出站消息。

Starting with version 5.0, the topic, qos, and retained properties are mapped to .RECEIVED_… headers (MqttHeaders.RECEIVED_TOPIC, MqttHeaders.RECEIVED_QOS, and MqttHeaders.RECEIVED_RETAINED), to avoid inadvertent propagation to an outbound message that (by default) uses the MqttHeaders.TOPIC, MqttHeaders.QOS, and MqttHeaders.RETAINED headers.

Adding and Removing Topics at Runtime

从版本 4.1 开始，您可以通过编程方式更改适配器订阅的主题。Spring Integration 提供 addTopic() 和 removeTopic() 方法。添加主题时，您可以选择指定 QoS（默认值：1）。您还可以通过向 <control-bus/> 发送适当的消息来修改主题，其中包含适当的负载，例如："myMqttAdapter.addTopic('foo', 1)"。

Starting with version 4.1, you can programmatically change the topics to which the adapter is subscribed. Spring Integration provides the addTopic() and removeTopic() methods. When adding topics, you can optionally specify the QoS (default: 1). You can also modify the topics by sending an appropriate message to a <control-bus/> with an appropriate payload — for example: "myMqttAdapter.addTopic('foo', 1)".

停止和启动适配器对主题列表没有影响（它不会恢复到配置中的原始设置）。这些更改不会保留在应用程序上下文的生命周期之外。一个新的应用程序上下文将恢复到已配置的设置。

Stopping and starting the adapter has no effect on the topic list (it does not revert to the original settings in the configuration). The changes are not retained beyond the life cycle of the application context. A new application context reverts to the configured settings.

在适配器停止（或与代理断开连接）时更改主题会在下次建立连接时生效。

Changing the topics while the adapter is stopped (or disconnected from the broker) takes effect the next time a connection is established.

Manual Acks

从版本 5.3 开始，您可以将 manualAcks 属性设置为 true。通常用于异步确认交付。当设置为 true 时，标头（IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK）会添加到消息中，其值为 SimpleAcknowledgment。您必须调用 acknowledge() 方法才能完成交付。有关更多信息，请参阅 IMqttClient setManualAcks() 和 messageArrivedComplete() 的 Javadoc。为了方便起见，提供了一个标头访问器：

Starting with version 5.3, you can set the manualAcks property to true. Often used to asynchronously acknowledge delivery. When set to true, header (IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK) is added to the message with the value being a SimpleAcknowledgment. You must invoke the acknowledge() method to complete the delivery. See the Javadocs for IMqttClient setManualAcks() and messageArrivedComplete() for more information. For convenience a header accessor is provided:

StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge();

从版本 5.2.11 开始，当消息转换器引发异常或从 MqttMessage 转换返回 null 时，MqttPahoMessageDrivenChannelAdapter 会向 errorChannel（如果提供）发送 ErrorMessage。否则，将此转换错误重新抛出到 MQTT 客户端回调中。

Starting with version 5.2.11, when the message converter throws an exception or returns null from the MqttMessage conversion, the MqttPahoMessageDrivenChannelAdapter sends an ErrorMessage into the errorChannel, if provided. Re-throws this conversion error otherwise into an MQTT client callback.

Configuring with Java Configuration

以下 Spring Boot 应用展示了如何通过 Java 配置来配置入站适配器的一个示例：

The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration:

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

        };
    }

}

Configuring with the Java DSL

下面的 Spring Boot 应用程序提供使用 Java DSL 配置入站适配器的示例：

The following Spring Boot application provides an example of configuring the inbound adapter with the 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();
    }

}

Outbound Channel Adapter

出站通道适配器由 MqttPahoMessageHandler 实现，该处理程序包装在 ConsumerEndpoint 中。为了方便起见，您可以使用命名空间对其进行配置。

The outbound channel adapter is implemented by the MqttPahoMessageHandler, which is wrapped in a ConsumerEndpoint. For convenience, you can configure it by using the namespace.

从版本 4.1 开始，适配器支持异步发送操作，避免在确认交付之前阻塞。您可以发出应用程序事件，以便在需要时使应用程序确认交付。

Starting with version 4.1, the adapter supports asynchronous send operations, avoiding blocking until the delivery is confirmed. You can emit application events to enable applications to confirm delivery if desired.

以下列表显示了出站通道适配器可用的属性：

The following listing shows the attributes available for an outbound channel adapter:

<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	The client ID.
2	The broker URL.
3	An `MqttMessageConverter` (optional). The default `DefaultPahoMessageConverter` recognizes the following headers: `mqtt_topic`: The topic to which the message will be sent `mqtt_retained`: `true` if the message is to be retained `mqtt_qos`: The quality of service
4	The client factory.
5	The default quality of service. It is used if no `mqtt_qos` header is found or the `qos-expression` returns `null`. It is not used if you supply a custom `converter`.
6	An expression to evaluate to determine the qos. The default is `headers[mqtt_qos]`.
7	The default value of the retained flag. It is used if no `mqtt_retained` header is found. It is not used if a custom `converter` is supplied.
8	An expression to evaluate to determine the retained boolean. The default is `headers[mqtt_retained]`.
9	The default topic to which the message is sent (used if no `mqtt_topic` header is found).
10	An expression to evaluate to determine the destination topic. The default is `headers['mqtt_topic']`.
11	When `true`, the caller does not block. Rather, it waits for delivery confirmation when a message is sent. The default is `false` (the send blocks until delivery is confirmed).
12	When `async` and `async-events` are both `true`, an `MqttMessageSentEvent` is emitted (See Events). It contains the message, the topic, the `messageId` generated by the client library, the `clientId`, and the `clientInstance` (incremented each time the client is connected). When the delivery is confirmed by the client library, an `MqttMessageDeliveredEvent` is emitted. It contains the `messageId`, the `clientId`, and the `clientInstance`, enabling delivery to be correlated with the `send()`. Any `ApplicationListener` or an event inbound channel adapter can receive these events. Note that it is possible for the `MqttMessageDeliveredEvent` to be received before the `MqttMessageSentEvent`. The default is `false`.

从 4.1 版开始，可以省略 URL。相反，服务器 URI 可以提供给 DefaultMqttPahoClientFactory 的 serverURIs 属性。例如，这支持连接到高可用性 (HA) 群集。

Starting with version 4.1, the URL can be omitted. Instead, the server URIs can be provided in the serverURIs property of the DefaultMqttPahoClientFactory. This enables, for example, connection to a highly available (HA) cluster.

Configuring with Java Configuration

以下 Spring Boot 应用程序展示了如何使用 Java 配置配置出站适配器：

The following Spring Boot application show an example of how to configure the outbound adapter with Java configuration:

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

    }

}

Configuring with the Java DSL

以下 Spring Boot 应用程序提供了一个使用 Java DSL 配置出站适配器的示例：

The following Spring Boot application provides an example of configuring the outbound adapter with the 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"));
    }

}

Events

适配器会发布某些应用程序事件。

Certain application events are published by the adapters.

MqttConnectionFailedEvent - published by both adapters if we fail to connect or a connection is subsequently lost. For the MQTT v5 Paho client, this event is also emitted when the server performs a normal disconnection, in which case the cause of the lost connection is null.
MqttMessageSentEvent - published by the outbound adapter when a message has been sent, if running in asynchronous mode.
MqttMessageDeliveredEvent - published by the outbound adapter when the client indicates that a message has been delivered, if running in asynchronous mode.
MqttSubscribedEvent - published by the inbound adapter after subscribing to the topics.

这些事件可以通过 ApplicationListener<MqttIntegrationEvent> 或带有 @EventListener 方法接收。

These events can be received by an ApplicationListener<MqttIntegrationEvent> or with an @EventListener method.

要确定事件的来源，请使用以下方法；您可以检查 bean 名称和/或连接选项（以访问服务器 URI 等）。

To determine the source of an event, use the following; you can check the bean name and/or the connect options (to access the server URIs etc).

MqttPahoComponent source = event.getSourceAsType();
String beanName = source.getBeanName();
MqttConnectOptions options = source.getConnectionInfo();

MQTT v5 Support

从版本 5.5.5 开始，spring-integration-mqtt 模块为 MQTT v5 协议提供通道适配器实现。org.eclipse.paho:org.eclipse.paho.mqttv5.client 是一个 optional 依赖项，因此必须明确包含在目标项目中。

Starting with version 5.5.5, the spring-integration-mqtt module provides channel adapter implementations for the MQTT v5 protocol. The org.eclipse.paho:org.eclipse.paho.mqttv5.client is an optional dependency, so has to be included explicitly in the target project.

由于 MQTT v5 协议支持 MQTT 消息中的额外任意属性，因此引入了 MqttHeaderMapper 实现来在发布和接收操作中进行映射。默认情况下（通过 * 模式），它映射所有接收的 PUBLISH 帧属性（包括用户属性）。在出站方面，它为 PUBLISH 帧映射标头的以下子集：contentType、mqtt_messageExpiryInterval、mqtt_responseTopic、mqtt_correlationData。

Since the MQTT v5 protocol supports extra arbitrary properties in an MQTT message, the MqttHeaderMapper implementation has been introduced to map to/from headers on publish and receive operations. By default, (via the * pattern) it maps all the received PUBLISH frame properties (including user properties). On the outbound side it maps this subset of headers for PUBLISH frame: 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 ConfigurableCompositeMessageConverter`bean。注意：仅当请求的消息有效载荷已经是 `MqttMessage`时，才不会使用提供的 `HeaderMapper<MqttProperties>。下面的 Java DSL 配置示例演示了如何在集成流中使用此通道适配器：

The outbound channel adapter for the MQTT v5 protocol is present as an Mqttv5PahoMessageHandler. It requires a clientId and MQTT broker URL or MqttConnectionOptions reference. It supports a MqttClientPersistence option, can be async and can emit MqttIntegrationEvent objects in that case (see asyncEvents option). If a request message payload is an org.eclipse.paho.mqttv5.common.MqttMessage, it is published as is via the internal IMqttAsyncClient. If the payload is byte[] it is used as is for the target MqttMessage payload to publish. If the payload is a String it is converted to byte[] to publish. The remaining use-cases are delegated to the provided MessageConverter which is a IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME ConfigurableCompositeMessageConverter bean from the application context. Note: the provided HeaderMapper<MqttProperties> is not used when the requested message payload is already an MqttMessage. The following Java DSL configuration sample demonstrates how to use this channel adapter in the integration flow:

@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 协议。

The org.springframework.integration.mqtt.support.MqttMessageConverter cannot be used with the Mqttv5PahoMessageHandler since its contract is aimed only for the MQTT v3 protocol.

如果连接在启动或运行时失败，则 Mqttv5PahoMessageHandler 会尝试在向此处理程序发送下一条消息时重新连接。如果此手动重新连接失败，则连接异常会抛回给调用者。在这种情况下，将应用标准的 Spring Integration 错误处理过程，包括请求处理程序建议，例如重试或断路器。

If connection fails on start up or at runtime, the Mqttv5PahoMessageHandler tries to reconnect on the next message produced to this handler. If this manual reconnection fails, the connection is exception is thrown back to the caller. In this case the standard Spring Integration error handling procedure is applied, including request handler advices, e.g. retry or circuit breaker.

在 Mqttv5PahoMessageHandler javadoc 及其超类中查看更多信息。

See more information in the Mqttv5PahoMessageHandler javadocs and its superclass.

MQTT v5 协议的传入通道适配器以 Mqttv5PahoMessageDrivenChannelAdapter`的形式出现。它需要一个 `clientId`以及 MQTT 代理 URL 或 `MqttConnectionOptions`引用，外加从中订阅和使用的主题。它支持一个 `MqttClientPersistence`选项，默认情况下为 in-memory。可以配置预期的 `payloadType（默认情况下为 byte[]），它传播到提供的 SmartMessageConverter`以从接收到的 `MqttMessage`的 `byte[]`中进行转换。如果设置了 `manualAck`选项，则会将 `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK`标头添加到要作为 `SimpleAcknowledgment`实例生成的消息中。`HeaderMapper<MqttProperties>`用于将 `PUBLISH`帧属性（包括用户属性）映射到目标消息标头。标准 `MqttMessage`属性（例如 `qos、id、dup、retained，以及接收的主题）始终映射到标头。有关更多信息，请参阅 MqttHeaders。

The inbound channel adapter for the MQTT v5 protocol is present as an Mqttv5PahoMessageDrivenChannelAdapter. It requires a clientId and MQTT broker URL or MqttConnectionOptions reference, plus topics to which to subscribe and consume from. It supports a MqttClientPersistence option, which is in-memory by default. The expected payloadType (byte[] by default) can be configured, and it is propagated to the provided SmartMessageConverter for conversion from byte[] of the received MqttMessage. If the manualAck option is set, then an IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK header is added to the message to produce as an instance of SimpleAcknowledgment. The HeaderMapper<MqttProperties> is used to map PUBLISH frame properties (including user properties) into the target message headers. Standard MqttMessage properties, such as qos, id, dup, retained, plus received topic are always mapped to headers. See MqttHeaders for more information.

从 6.3 版开始，Mqttv5PahoMessageDrivenChannelAdapter 基于 MqttSubscription 提供了构造函数，用于细粒度配置而不是普通主题名称。当提供这些订阅时，信道适配器的 qos 选项不可用，因为此 qos 模式是 MqttSubscription API 的一部分。

Starting with version 6.3, the Mqttv5PahoMessageDrivenChannelAdapter provides constructors based on the MqttSubscription for fine-grained configuration instead of plain topic names. When these subscriptions are provided, the qos option of the channel adapter cannot be used, since such a qos mode is a part of MqttSubscription API.

以下 Java DSL 配置样本演示了如何在集成流中使用此信道适配器：

The following Java DSL configuration sample demonstrates how to use this channel adapter in the integration flow:

@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 协议。

The org.springframework.integration.mqtt.support.MqttMessageConverter cannot be used with the Mqttv5PahoMessageDrivenChannelAdapter since its contract is aimed only for the MQTT v3 protocol.

在 Mqttv5PahoMessageDrivenChannelAdapter javadoc 及其超类中查看更多信息。

See more information in the Mqttv5PahoMessageDrivenChannelAdapter javadocs and its superclass.

建议将 MqttConnectionOptions#setAutomaticReconnect(boolean) 设置为 true，以让内部 IMqttAsyncClient 实例处理重新连接。否则，只有手动的 Mqttv5PahoMessageDrivenChannelAdapter 重新启动才能处理重新连接，例如通过在断开连接时处理 MqttConnectionFailedEvent。

It is recommended to have the MqttConnectionOptions#setAutomaticReconnect(boolean) set to true to let an internal IMqttAsyncClient instance to handle reconnects. Otherwise, only the manual restart of Mqttv5PahoMessageDrivenChannelAdapter can handle reconnects, e.g. via MqttConnectionFailedEvent handling on disconnection.

Shared MQTT Client Support

如果多个集成需要一个 MQTT ClientID，不能使用多个 MQTT 客户端实例，因为 MQTT 代理可能对每个 ClientID 的连接数有限制（通常，允许单个连接）。要将一个客户端重复用于不同的信道适配器，可以使用 org.springframework.integration.mqtt.core.ClientManager 组件并将其传递给任何需要的信道适配器。它将管理 MQTT 连接生命周期并且在需要时自动重新连接。此外，可以向客户端管理器提供自定义连接选项和 MqttClientPersistence，就像当前可以对信道适配器组件执行相同操作一样。

If a single MQTT ClientID is required for several integrations, multiple MQTT client instances cannot be used because MQTT brokers may have a limitation on a number of connections per ClientID (typically, a single connection is allowed). For having a single client reused for different channel adapters, a org.springframework.integration.mqtt.core.ClientManager component may be used and passed to any channel adapter needed. It will manage MQTT connection lifecycle and do automatic reconnects if needed. Also, a custom connection options and MqttClientPersistence may be provided to the client manager just as currently it can be done for channel adapter components.

请注意，支持 MQTT v5 和 v3 信道适配器。

Note that both MQTT v5 and v3 channel adapters are supported.

以下 Java DSL 配置样本演示了如何在集成流中使用此客户端管理器：

The following Java DSL configuration sample demonstrates how to use this client manager in the integration flow:

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