JMS Support

Outbound Gateway, 出站网关, reply-destination, reply-destination-name, reply-channe, async, idle-reply-listener-timeout, reply-listern, correlation-key, JMSMessageID, JMSCorrelationID, jms_correlationId

Spring Integration 为接收和发送 JMS 消息提供通道适配器。你需要将此依赖项包含在你的项目中：

Maven
Gradle

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

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

必须通过某些 JMS 厂商特定的实现明确添加“jakarta.jms:jakarta.jms-api”，例如 Apache ActiveMQ。实际上有两个基于 JMS 的入站通道适配器。第一个基于轮询周期使用 Spring 的“JmsTemplate”来接收。第二个为“message-driven”，并依赖于 Spring 的“MessageListener”容器。出站通道适配器使用“JmsTemplate”按需转换并发送一个 JMS 消息。通过使用 JmsTemplate 和 MessageListener 容器，Spring Integration 依赖于 Spring 的 JMS 支持。这一点很重要，因为在这些适配器上公开的大多数属性都会配置底层 JmsTemplate 和 MessageListener 容器。有关 JmsTemplate 和 MessageListener 容器的更多详细信息，请参阅 Spring JMS documentation。虽然 JMS 通道适配器专用于单向消息传递（仅发送或仅接收），但 Spring Integration 也提供用于请求和答复操作的入站和出站 JMS 网关。入站网关依赖于 Spring 的 MessageListener 容器实现之一进行消息驱动的接收。它还能够将返回值发送到 reply-to 目标，如接收到的消息所提供。出站网关向 request-destination（或 request-destination-name 或 request-destination-expression）发送 JMS 消息，然后接收答复消息。您可以显式配置 reply-destination 引用（或 reply-destination-name 或 reply-destination-expression）。否则，出站网关将使用 JMS TemporaryQueue。在 Spring Integration 2.2 之前，如果有必要，会为每个请求或答复创建一个“TemporaryQueue”（并将其移除）。从 Spring Integration 2.2 开始，您可以配置出站网关以使用“MessageListener”容器来接收答复，而不是直接使用一个新的（或缓存的）“Consumer”来接收每个请求的答复。如果这样配置且没有提供明确的答复目标，则会为每个网关使用一个“TemporaryQueue”，而不是为每个请求使用一个。从版本 6.0 开始，如果将“replyPubSubDomain”选项设置为“true”，出站网关将创建一个“TemporaryTopic”而不是“TemporaryQueue”。一些 JMS 厂商以不同方式处理这些目标。

Inbound Channel Adapter

入站通道适配器需要一个指向一个“JmsTemplate”实例的引用或指向一个“ConnectionFactory”和一个“Destination”的引用（您可以提供“destinationName”来替代“destination”引用）。以下示例定义了一个具有“Destination”引用的入站通道适配器：

Java DSL
Kotlin DSL
Java
XML

@Bean
public IntegrationFlow jmsInbound(ConnectionFactory connectionFactory) {
    return IntegrationFlow.from(
                    Jms.inboundAdapter(connectionFactory)
                       .destination("inQueue"),
                    e -> e.poller(poller -> poller.fixedRate(30000)))
            .handle(m -> System.out.println(m.getPayload()))
            .get();
}

@Bean
fun jmsInbound(connectionFactory: ConnectionFactory) =
    integrationFlow(
            Jms.inboundAdapter(connectionFactory).destination("inQueue"),
            { poller { Pollers.fixedRate(30000) } })
       {
            handle { m -> println(m.payload) }
       }

@Bean
@InboundChannelAdapter(value = "exampleChannel", poller = @Poller(fixedRate = "30000"))
public MessageSource<Object> jmsIn(ConnectionFactory connectionFactory) {
    JmsDestinationPollingSource source = new JmsDestinationPollingSource(new JmsTemplate(connectionFactory));
    source.setDestinationName("inQueue");
    return source;
}

<int-jms:inbound-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel">
    <int:poller fixed-rate="30000"/>
</int-jms:inbound-channel-adapter>

从前面的配置中注意到，通道适配器是一个轮询消费者。这意味着它会在被触发时调用消费器。你应该只在轮询相对不频繁且及时性不重要的情况下使用它。对于所有其他情况（绝大多数基于 JMS 的用例），消息驱动通道适配器（MDBA）是更好的选择。

默认情况下，所有需要引用 JMS 连接工厂的 JMS 适配器都会自动查找名为“jmsConnectionFactory”的 bean。这就是为什么在许多示例中看不到 connectionFactory 属性的原因。但是，如果你的 JMS 连接工厂具有不同的 bean 名称，则需要提供该属性。

如果“extract-payload”设置为“true”（默认值），接收的 JMS 消息会通过“MessageConverter”。当依靠默认的“SimpleMessageConverter”时，这意味着生成的 Spring Integration 消息具有 JMS 消息的主体作为其负载。一个 JMS“TextMessage”生成一个基于字符串的负载，一个 JMS“BytesMessage”生成一个字节数组负载，一个 JMS“ObjectMessage”的可序列化实例成为 Spring Integration 消息的负载。如果您希望原始 JMS 消息作为 Spring Integration 消息的负载，请将“extractPayload”选项设置为“false”。

从版本 5.0.8 开始，“receive-timeout”的默认值为“-1”（不等待）用于“org.springframework.jms.connection.CachingConnectionFactory”和“cacheConsumers”，否则为 1 秒。JMS 入站通道适配器基于提供的“ConnectionFactory”和选项创建一个“DynamicJmsTemplate”。如果需要外部“JmsTemplate”（例如在 Spring Boot 环境中）或“ConnectionFactory”没有缓存或没有“cacheConsumers”，如果期望无阻塞消耗，建议设置“jmsTemplate.receiveTimeout(-1)”：

Jms.inboundAdapter(connectionFactory)
        .destination(queueName)
        .configureJmsTemplate(template -> template.receiveTimeout(-1))

Transactions

从版本 4.0 开始，入站通道适配器支持“session-transacted”属性。在早期版本中，您必须注入一个“sessionTransacted”设置为“true”的“JmsTemplate”。（适配器允许您将“acknowledge”属性设置为“transacted”，但这是不正确的，也不起作用）。

但是，请注意，将“session-transacted”设置为“true”没什么用，因为事务会在“receive()”操作后并消息被发送到“channel”前立即提交。

如果您想要整个流程成为事务性的（例如，如果有下游出站通道适配器），则必须使用具有“JmsTransactionManager”的事务性轮询器。或者，考虑使用“acknowledge”设置为“transacted”（默认值）的“jms-message-driven-channel-adapter”。

Message Driven Channel Adapter

“message-driven-channel-adapter”需要一个指向 Spring 的“MessageListener”容器（“AbstractMessageListenerContainer”的任何子类）实例的引用或指向“ConnectionFactory”和“Destination”的引用（可以提供“destinationName”来替代“destination”引用）。以下示例定义了一个具有“Destination”引用的消息驱动通道适配器：

Java DSL
Kotlin DSL
Java
XML

@Bean
public IntegrationFlow jmsMessageDrivenRedeliveryFlow() {
    return IntegrationFlow
            .from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory())
                     .destination("inQueue"))
            .channel("exampleChannel")
            .get();
}

@Bean
fun jmsMessageDrivenFlowWithContainer() =
        integrationFlow(
                Jms.messageDrivenChannelAdapter(jmsConnectionFactory())
                             .destination("inQueue")) {
            channel("exampleChannel")
        }

@Bean
public JmsMessageDrivenEndpoint jmsIn() {
    JmsMessageDrivenEndpoint endpoint = new JmsMessageDrivenEndpoint(container(), listener());
    return endpoint;
}
@Bean
public AbstractMessageListenerContainer container() {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(cf());
    container.setDestinationName("inQueue");
    return container;
}

@Bean
public ChannelPublishingJmsMessageListener listener() {
    ChannelPublishingJmsMessageListener listener = new ChannelPublishingJmsMessageListener();
    listener.setRequestChannelName("exampleChannel");
    return listener;
}

<int-jms:message-driven-channel-adapter id="jmsIn" destination="inQueue" channel="exampleChannel"/>

消息驱动的适配器还接受与 MessageListener 容器相关的几个属性。仅当您不提供 container 引用时，才会考虑这些值。在这种情况下，将基于这些属性创建一个 DefaultMessageListenerContainer 实例并将其配置好。例如，您可以指定 transaction-manager 引用、concurrent-consumers 值以及几个其他属性引用和值。请参阅 Javadoc 和 Spring Integration 的 JMS 架构 (spring-integration-jms.xsd) 以获取更多详细信息。如果您有自定义侦听器容器实现（通常是“DefaultMessageListenerContainer”的子类），您可以通过使用“container”属性提供对它的实例的引用或通过使用“container-class”属性提供它的完全限定类名。在这种情况下，适配器上的属性会传输到自定义容器的实例。

您不能使用 Spring JMS 命名空间元素“<jms:listener-container/>”来配置“<int-jms:message-driven-channel-adapter>”的容器引用，因为该元素实际上并不引用容器。每个“<jms:listener/>”子元素都有它自己的“DefaultMessageListenerContainer”（具有在父元素“<jms:listener-container/>”上定义的共享属性）。不过，您可以给每个侦听器子元素一个“id”，并用它注入通道适配器，“<jms:/>”命名空间需要一个真正的侦听器。建议为“DefaultMessageListenerContainer”配置一个常规的“<bean>”，并在通道适配器中用它作引用。

从 4.2 版本开始，默认 message-driven 模式为 auto，除非您提供了一个外部容器。在后一种情况下，您应该根据需要配置容器。我们建议使用 pub-sub，因为只有这样才能避免丢失消息。

“extract-payload”属性具有相同的效果，其默认值为“true”。由于“poller”元素会被主动调用，因此它不适用于消息驱动通道适配器。对于大多数方案，消息驱动方法更好，因为消息在从底层的 JMS 消费者接收到后会立即传递给“MessageChannel”。

最后，<message-driven-channel-adapter> 元素还接受“error-channel”属性。这提供了与 Enter the GatewayProxyFactoryBean 中描述相同的基本功能。以下示例演示了如何在消息驱动的通道适配器上设置错误通道：

<int-jms:message-driven-channel-adapter id="jmsIn" destination="inQueue"
    channel="exampleChannel"
    error-channel="exampleErrorChannel"/>

将前面该示例与我们后面讨论的通用网关配置或 JMS“inbound-gateway”进行比较时，主要区别在于这是一个单向流，因为它是一个“channel-adapter”，而不是网关。因此，“error-channel”下游流也应是单向的。例如，它可以发送到日志记录处理程序，或者可以连接到其他 JMS“<outbound-channel-adapter>”元素。

从主题使用时，将“pub-sub-domain”属性设置为 true。为持久订阅设置“subscription-durable”为“true”，或为共享订阅设置“subscription-shared” (它需要 JMS 2.0 代理并且从版本 4.2 开始可用)。使用“subscription-name”来命名订阅。

从版本 5.1 开始，当应用程序正在运行时停止端点时，基础监听器容器关闭，关闭其共享连接和使用者。以前，连接和使用者保持打开状态。要恢复到以前的行为，将 JmsMessageDrivenEndpoint 中的 shutdownContainerOnStop 设置为“false”。

从版本 6.3 开始，“ChannelPublishingJmsMessageListener”现在可以提供一个 RetryTemplate 和 RecoveryCallback<Message<?>>，可以在下行发送和发送接收操作中重试。这些选项也公开到 Java DSL 的 JmsMessageDrivenChannelAdapterSpec 中。

Inbound Conversion Errors

从版本 4.2 开始，“error-channel”也用于转换错误。以前，如果 JMS“<message-driven-channel-adapter/>”或“<inbound-gateway/>”由于转换错误而无法传递消息，则会向容器抛回异常。如果容器配置为使用事务，则消息将被回滚并重复传送。转换过程发生在消息构建之前和期间，以使此类错误不会发送到“error-channel”。现在，此类转换异常会导致向“error-channel”发送 ErrorMessage，异常作为“payload”。如果希望事务回滚并且定义了“error-channel”，则“error-channel”上的集成流必须重新抛出异常（或其他异常）。如果错误流不抛出异常，则事务将提交，消息将被移除。如果没有定义“error-channel”，则会向容器抛回异常，就像以前一样。

Outbound Channel Adapter

JmsSendingMessageHandler 实现 MessageHandler 接口，能够将 Spring Integration 消息`转换为 JMS 消息，然后发送到 JMS 目的地。它需要 `jmsTemplate 引用或 jmsConnectionFactory 和 destination 引用（destinationName 可以代替 destination 提供）。与入站通道适配器一样，配置此适配器的最简单方法是使用名称空间支持。以下配置生成了一个适配器，该适配器从 exampleChannel 接收 Spring Integration 消息，将它们转换为 JMS 消息，并将它们发送到 bean 名称是 outQueue 的 JMS 目的地引用：

Java DSL
Kotlin DSL
Groovy DSL
Java
XML

@Bean
public IntegrationFlow jmsOutboundFlow() {
    return IntegrationFlow.from("exampleChannel")
                .handle(Jms.outboundAdapter(cachingConnectionFactory())
                    .destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)
                    .configureJmsTemplate(t -> t.id("jmsOutboundFlowTemplate")));
}

@Bean
fun jmsOutboundFlow() =
        integrationFlow("exampleChannel") {
            handle(Jms.outboundAdapter(jmsConnectionFactory())
                    .apply {
                        destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)
                        deliveryModeFunction<Any> { DeliveryMode.NON_PERSISTENT }
                        timeToLiveExpression("10000")
                        configureJmsTemplate { it.explicitQosEnabled(true) }
                    }
            )
        }

@Bean
jmsOutboundFlow() {
    integrationFlow('exampleChannel') {
        handle(Jms.outboundAdapter(new ActiveMQConnectionFactory())
                .with {
                    destinationExpression 'headers.' + SimpMessageHeaderAccessor.DESTINATION_HEADER
                    deliveryModeFunction { DeliveryMode.NON_PERSISTENT }
                    timeToLiveExpression '10000'
                    configureJmsTemplate {
                        it.explicitQosEnabled true
                    }
                }
        )
    }
}

@Bean
@ServiceActivator(inputChannel = "exampleChannel")
public MessageHandler jmsOut() {
    JmsSendingMessageHandler handler = new JmsSendingMessageHandler(new JmsTemplate(connectionFactory));
    handler.setDestinationName("outQueue");
    return handler;
}

<int-jms:outbound-channel-adapter id="jmsOut" destination="outQueue" channel="exampleChannel"/>

与入站通道适配器一样，有一个“extract-payload”属性。但是，对于出站适配器，含义是相反的。布尔属性不是应用于 JMS 消息，而是应用于 Spring Integration 消息负载。换句话说，决定是将 Spring Integration 消息本身作为 JMS 消息正文传递，还是将 Spring Integration 消息负载作为 JMS 消息正文传递。默认值为“true”。因此，如果你传递一个负载为 String 的 Spring Integration 消息，将创建一个 JMS TextMessage。另一方面，如果你想通过 JMS 将实际的 Spring Integration 消息发送到另一个系统，请将其设置为“false”。

无论有效负载提取的布尔值是什么，Spring Integration 都会将属性映射到 JMS 属性，只要您依赖于默认转换器或提供对其他 MessageConverter 实例的引用。（对于“入站”适配器也是如此，只是在这种情况下，JMS 属性映射到 Spring Integration 属性）。

从版本 5.1 开始，“<int-jms:outbound-channel-adapter>” (JmsSendingMessageHandler) 可以使用 deliveryModeExpression 和 timeToLiveExpression 属性来评估弹性消息中间件针对请求 Spring 消息 发送在运行时适用的 QoS 值。DefaultJmsHeaderMapper 的新 setMapInboundDeliveryMode(true) 和 setMapInboundExpiration(true) 选项可以作为消息头中动态 deliveryMode 和 timeToLive 信息的来源：

<int-jms:outbound-channel-adapter delivery-mode-expression="headers.jms_deliveryMode"
                        time-to-live-expression="headers.jms_expiration - T(System).currentTimeMillis()"/>

Transactions

从版本 4.0 开始，出站通道适配器支持 session-transacted 属性。在较早的版本中，你必须注入一个 JmsTemplate，并将 sessionTransacted 设置为“true”。该属性现在在内置默认 JmsTemplate 上设置属性。如果存在事务（可能来自上游 message-driven-channel-adapter），则发送操作在同一事务内执行。否则，将启动一个新事务。

Inbound Gateway

Spring Integration 的消息驱动 JMS 入站网关委托给一个 MessageListener 容器，支持动态调整并发使用者，并且还可以处理应答。入站网关需要一个 ConnectionFactory 和一个请求 Destination（或“requestDestinationName”）的引用。以下示例定义了一个 JMS inbound-gateway，它从 bean id inQueue 引用的 JMS 队列中接收，并发送到名为 exampleChannel 的 Spring Integration 通道：

<int-jms:inbound-gateway id="jmsInGateway"
    request-destination="inQueue"
    request-channel="exampleChannel"/>

鉴于网关提供的是请求-答复行为，而不是单向发送或接收行为，所以它们还具有两个针对 “payload extraction” 的明显属性（如：用于通道适配器“extract-payload”设置的 discussed earlier）。对于入站网关，“extract-request-payload”属性确定是否提取接收到的 JMS 消息正文。如果为“false”，则 JMS 消息本身将成为 Spring Integration 消息有效负载。默认值为“true”。

类似地，对于入站网关，“extract-reply-payload”属性应用于要转换为应答 JMS 消息的 Spring Integration 消息。如果你想传递整个 Spring Integration 消息（作为 JMS ObjectMessage 的正文），请将此值设置为“false”。默认情况下，Spring Integration 消息负载也会转换为 JMS 消息（例如，String 负载将变为 JMS TextMessage）。

与其他任何事情一样，网关调用可能会导致错误。默认情况下，不会向生产者通知可能发生在消费者端的错误，并且会超时等待应答。但是，有时你可能希望将错误情况传达给消费者（换句话说，你可能希望通过将其映射到消息来将异常视为有效应答）。为了实现这一点，JMS 入站网关为消息通道提供支持，可以将错误发送到该通道以进行处理，这可能会导致符合某种约定的应答消息负载，该约定定义了调用方可能期待的“error”应答。你可以使用 error-channel 属性来配置这样的通道，如下例所示：

<int-jms:inbound-gateway request-destination="requestQueue"
          request-channel="jmsInputChannel"
          error-channel="errorTransformationChannel"/>

<int:transformer input-channel="exceptionTransformationChannel"
        ref="exceptionTransformer" method="createErrorResponse"/>

您可能会注意到，此示例看起来与包含在 Enter the GatewayProxyFactoryBean 中的示例非常相似。这里应用了相同的概念：exceptionTransformer 可以是创建错误响应对象的 POJO，您可以引用 nullChannel 来抑制错误，也可以将“error-channel”留空以让异常传播。

请参阅 Inbound Conversion Errors。

从 4.2 版本开始，默认“message-driven”模式为“auto”，除非提供了外部容器。在这种情况下，您应该根据需要配置容器。我们建议使用“pub-sub”模式，因为只有这样才能避免丢失消息。

从版本 5.1 开始，当应用程序正在运行时停止端点时，基础监听器容器关闭，关闭其共享连接和使用者。以前，连接和使用者保持打开状态。要恢复到以前的行为，将 JmsInboundGateway 中的 shutdownContainerOnStop 设置为“false”。

默认情况下，JmsInboundGateway 在接收到的消息中查找 jakarta.jms.Message.getJMSReplyTo() 属性来确定将应答发送到哪里。否则，它可以配置为静态 defaultReplyDestination 或 defaultReplyQueueName 或 defaultReplyTopicName。此外，从版本 6.1 开始，可以在提供的 ChannelPublishingJmsMessageListener 上配置一个 replyToExpression 以动态确定应答目的地，如果请求上标准 JMSReplyTo 属性为 null。接收到的 jakarta.jms.Message 用作根评估上下文对象。以下示例演示了如何使用 Java DSL API 来配置一个入站 JMS 网关，该网关具有从请求消息解析的自定义应答目的地：

@Bean
public IntegrationFlow jmsInboundGatewayFlow(ConnectionFactory connectionFactory) {
    return IntegrationFlow.from(
                    Jms.inboundGateway(connectionFactory)
                            .requestDestination("requestDestination")
                            .replyToFunction(message -> message.getStringProperty("myReplyTo")))
            .<String, String>transform(String::toUpperCase)
            .get();
}

从 6.3 版本开始，Jms.inboundGateway() API 公开了 retryTemplate() 和 recoveryCallback() 选项，用于重试内部收发操作。

Outbound Gateway

出站网关使用 Spring Integration 消息创建 JMS 消息，并将其发送到 request-destination。然后，通过使用选择器从配置的 reply-destination 接收或（如果未提供 reply-destination）通过创建 JMS TemporaryQueue（或如果 replyPubSubDomain= true，则创建 TemporaryTopic）实例来处理 JMS 答复消息。

将 reply-destination（或 reply-destination-name）与将 cacheConsumers 设置为 true 的 CachingConnectionFactory 配合使用会导致出现内存不足的情况。这是因为每个请求都会获得带有新选择器的消费者（选择基于 correlation-key 值，或当没有 correlation-key 时，基于发送的 JMSMessageID）。鉴于这些选择器是唯一的，因此它们在当前请求完成后会保留在缓存中（未使用）。如果您指定了答复目标，建议您不要使用缓存使用者。或者，考虑作为 described below 使用 <reply-listener/>。

以下示例显示了如何配置出站网关：

<int-jms:outbound-gateway id="jmsOutGateway"
    request-destination="outQueue"
    request-channel="outboundJmsRequests"
    reply-channel="jmsReplies"/>

“outbound-gateway”有效负载提取属性与“inbound-gateway” 的属性相反（请参阅 earlier discussion）。这意味着“extract-request-payload”属性值适用于转换为要作为请求发送的 JMS 消息的 Spring Integration 消息。“extract-reply-payload”属性值适用于作为答复接收的 JMS 消息，然后转换为 Spring Integration 消息，以便随后发送到“reply-channel”，如前面的配置示例所示。

Using a `<reply-listener/>`

Spring Integration 2.2 引入了一种处理响应的备用技术。如果您向网关添加 <reply-listener/> 子元素而不是为每个响应创建使用者，那么 MessageListener 容器将用于接收响应并将其移交给请求线程。这提供了许多性能优势，还缓解了 earlier caution 中描述的缓存使用者内存使用问题。

当将 <reply-listener/> 与没有 reply-destination 的出站网关一起使用时，它会为每个请求使用一个 TemporaryQueue，而不是为每个请求创建一个 TemporaryQueue。（如果与代理断开连接再恢复，网关会根据需要创建额外的 TemporaryQueue）。如果将 replyPubSubDomain 设置为 true，则从 6.0 版本开始，会创建一个 TemporaryTopic。

在使用 correlation-key 时，多个网关可以共享同一个答复目标，因为监听器容器使用适用于每个网关的唯一选择器。

如果你指定了答复监听器和答复目标（或答复目标名称）但不提供关联键，网关会记录警告并回退到 2.2 版本之前的行为。这是因为在这种情况下无法配置选择器。因此，没法避免答复转到可能配置相同答复目标的不同网关。请注意，在这种情况下，每个请求都会使用一个新的消费者，并且消费者会像上面的警告中所述的那样在内存中累积；因此，在这种情况下不应使用缓存的消费者。

以下示例显示了具有默认属性的答复监听器：

<int-jms:outbound-gateway id="jmsOutGateway"
        request-destination="outQueue"
        request-channel="outboundJmsRequests"
        reply-channel="jmsReplies">
    <int-jms:reply-listener />
</int-jms-outbound-gateway>

侦听器非常轻量级，我们预计在大多数情况下只需要一个使用者。但是，您可以添加诸如 concurrent-consumers、max-concurrent-consumers 等属性。参阅模式了解支持属性的完整列表，以及它们含义的 Spring JMS documentation。

Idle Reply Listeners

从 4.2 版本开始，您可以根据需要启动回复侦听器（并在空闲时间后停止侦听器），而不是在网关生命周期持续时间内运行侦听器。如果应用程序上下文中有很多大部分处于空闲状态的网关，这将非常有用。这种情况之一是很多（非活动）分区 Spring Batch 作业使用 Spring Integration 和 JMS 进行分区分布时的上下文。如果所有答复侦听器都处于活动状态，则 JMS 服务器对每个网关都有一个活动使用者。通过启用空闲超时，每个使用者只在相应批处理作业运行时（以及作业完成后的一段时间内）存在。

请参阅 Attribute Reference 中的 idle-reply-listener-timeout。

Gateway Reply Correlation

本节根据网关的配置方式描述了用于回复关联的机制（确保原始网关仅接收对自身请求的回复）。请参阅 Attribute Reference 完整了解此处讨论的属性。

以下列表描述了各种场景（数字仅用于标识——顺序无关紧要）：

没有 reply-destination*`属性，也没有 `<reply-listener>`为每个请求创建一个 `TemporaryQueue，并在请求完成（成功或失败）时删除它。correlation-key 无关紧要。
提供了一个 reply-destination*`属性，并且既没有提供一个 `<reply-listener/>，也没有提供一个 correlation-key`将 `JMSCorrelationID 等于传出消息用作消费者的消息选择器： messageSelector = "JMSCorrelationID = '" + 消息 ID + "'" 响应系统预期在回复 JMSCorrelationID 中返回输入 JMSMessageID。这是一个常见模式，并已由 Spring Integration 入站网关以及 Spring 的 MessageListenerAdapter 为消息驱动的 POJO 实现。

当你使用此配置时，不应该将主题用于答复。答复可能会丢失。

提供了一个 reply-destination*`属性，没有提供 `<reply-listener/>，并且 correlation-key="JMSCorrelationID"`该网关生成一个唯一的关联 ID 并将其插入到 `JMSCorrelationID 头中。消息选择器： messageSelector = "JMSCorrelationID = '" + 唯一 ID + "'" 响应系统预期在回复 JMSCorrelationID 中返回输入 JMSCorrelationID。这是一个常见模式，并已由 Spring Integration 入站网关以及 Spring 的 MessageListenerAdapter 为消息驱动的 POJO 实现。
提供了一个 reply-destination*`属性，没有提供 `<reply-listener/>，并且 correlation-key="myCorrelationHeader"`该网关生成一个唯一的关联 ID 并将其插入到 `myCorrelationHeader 消息属性中。correlation-key 可以是任意用户定义的值。消息选择器： messageSelector = "myCorrelationHeader = '" + 唯一 ID + "'" 响应系统预期在回复 myCorrelationHeader 中返回输入 myCorrelationHeader。
提供了一个 reply-destination*`属性，没有提供 `<reply-listener/>，并且 correlation-key="JMSCorrelationID*"（注意相关键中的 *）。该网关使用请求消息（如果存在）中 jms_correlationId 头中的值，并将其插入到 JMSCorrelationID 头中。消息选择器： messageSelector = "JMSCorrelationID = '" + headers['jms_correlationId'] + "'" 用户必须确保该值唯一。如果该标头不存在，该网关将表现为 3。响应系统预期在回复 JMSCorrelationID 中返回输入 JMSCorrelationID。这是一个常见模式，并已由 Spring Integration 入站网关以及 Spring 的 MessageListenerAdapter 为消息驱动的 POJO 实现。
没有提供 reply-destination*`属性，并且提供了一个 `<reply-listener>`创建一个临时队列，并将其用于此网关实例的所有回复。消息中不需要关联数据，但输出 `JMSMessageID 在网关中内部使用，将回复定向到正确的请求线程。
提供了一个 reply-destination*`属性，提供了一个 `<reply-listener>，并且没有提供 correlation-key`不允许。 \<reply-listener/\> 配置被忽略，并且网关表现为 `2。会写入一条警告日志消息来表示此情况。
提供了一个 reply-destination*`属性，提供了一个 `<reply-listener>，并且 correlation-key="JMSCorrelationID"`该网关具有一个唯一的关联 ID，并将其与 `JMSCorrelationID 头中的增量值一起插入（gatewayId + "_" + ++seq）。消息选择器： messageSelector = "JMSCorrelationID LIKE '" + gatewayId%'" 响应系统预期在回复 JMSCorrelationID 中返回输入 JMSCorrelationID。这是一个常见模式，并已由 Spring Integration 入站网关以及 Spring 的 MessageListenerAdapter 为消息驱动的 POJO 实现。由于每个网关都有一个唯一 ID，每个实例只会收到自己的回复。所有关联数据都用于将回复路由到正确的请求线程。
提供了一个 reply-destination*`属性，提供了一个 `<reply-listener/>，并且 correlation-key="myCorrelationHeader"`该网关具有一个唯一的关联 ID，并将其与 `myCorrelationHeader 属性中的增量值一起插入（gatewayId + "_" + ++seq）。correlation-key 可以是任意用户定义的值。消息选择器： messageSelector = "myCorrelationHeader LIKE '" + gatewayId%'" 预期的响应系统在响应 myCorrelationHeader 中返回入站 myCorrelationHeader 。由于每个网关都有一个唯一的 ID，所以每个实例仅获取其自己的响应。完整的关联数据用于将 reply 路由到正确的请求线程。
提供了一个 reply-destination*`属性，提供了一个 `<reply-listener/>，并且 correlation-key="JMSCorrelationID*"（注意关联键中的 * ）不允许。使用响应侦听器时，不允许用户提供的关联 ID。网关不会使用此配置初始化。

Async Gateway

从版本 4.3 开始，现在可以在配置出站网关时指定 async="true"（或在 Java 中 setAsync(true)）。

默认情况下，当向网关发送请求时，请求线程将挂起，直到收到回复。然后流继续在该线程上。如果 async 为 true，则请求线程在 send() 完成后立即释放，并在侦听器容器线程上返回回复（并继续流）。当网关在轮询线程上调用时，这可能很有用。线程被释放，可用于框架内的其他任务。

async 需要一个 <reply-listener/> （或在使用 Java 配置时 setUseReplyContainer(true) ）。它还需要一个要指定的 correlationKey （通常是 JMSCorrelationID ）。如果未满足以上任一条件，则忽略 async 。

Attribute Reference

以下列表显示 outbound-gateway 的所有可用属性：

<int-jms:outbound-gateway
    connection-factory="connectionFactory" 1
    correlation-key="" 2
    delivery-persistent="" 3
    destination-resolver="" 4
    explicit-qos-enabled="" 5
    extract-reply-payload="true" 6
    extract-request-payload="true" 7
    header-mapper="" 8
    message-converter="" 9
    priority="" 10
    receive-timeout="" 11
    reply-channel="" 12
    reply-destination="" 13
    reply-destination-expression="" 14
    reply-destination-name="" 15
    reply-pub-sub-domain="" 16
    reply-timeout="" 17
    request-channel="" 18
    request-destination="" 19
    request-destination-expression="" 20
    request-destination-name="" 21
    request-pub-sub-domain="" 22
    time-to-live="" 23
    requires-reply="" 24
    idle-reply-listener-timeout="" 25
    async=""> 26
  <int-jms:reply-listener /> 27
</int-jms:outbound-gateway>

1 引用一个 jakarta.jms.ConnectionFactory。默认 jmsConnectionFactory。

2 一个属性的名称，其中包含将响应与回复相关联的相关数据。如果省略，网关希望响应系统在 JMSCorrelationID`标头中返回出站 `JMSMessageID`标头。如果指定，网关会生成一个关联 ID，并将指定属性填充进去。响应系统必须在同一属性中回显该值。它可以设置为 `JMSCorrelationID，在这种情况下，标准标头（而不是 String`属性）用于保存关联数据。当你使用 `<reply-container/>`时，如果你提供了一个显式的 `reply-destination，你必须指定 correlation-key。从 4.0.1 版本开始，此属性还支持值 JMSCorrelationID*，这意味着如果出站消息已经有一个 JMSCorrelationID（从 `jms_correlationId`映射而来）标头，则使用该标头，而不是生成一个新的标头。请注意，当你使用 `<reply-container/>`时，不允许使用 `JMSCorrelationID*`键，因为容器需要在初始化期间设置消息选择器。====== 你应该明白，网关无法确保唯一性，如果提供的关联 ID 不是唯一的，则可能会产生意外的副作用。

1	一个布尔值，指示传递模式应该是 `DeliveryMode.PERSISTENT` (`true`) 还是 `DeliveryMode.NON_PERSISTENT` (`false`)。此设置仅在 `explicit-qos-enabled`为 `true`时生效。
2	一个 `DestinationResolver`。默认值是一个 `DynamicDestinationResolver`，它将目标名称映射到同名的队列或主题。
3	将其设置为 true`时，它启用使用服务质量属性： `priority、delivery-mode`和 `time-to-live。
4	当设置为 `true`（默认值）时，Spring Integration 回复消息的有效载荷通过使用 `MessageConverter`从 JMS 回复消息正文创建而来）。当设置为 `false`时，整个 JMS 消息都成为 Spring Integration 消息的有效载荷。
5	当设置为 `true`（默认值）时，Spring Integration 消息的有效载荷会转换为一个 `JMSMessage`（通过使用 `MessageConverter`）。当设置为 false`时，整个 Spring Integration 消息都转换为 `JMSMessage。在这两种情况下，Spring Integration 消息标头都通过使用 `HeaderMapper`映射到 JMS 标头和属性。
6	一个 `HeaderMapper`用于将 Spring Integration 消息标头和 JMS 消息标头及属性相互映射。
7	一个 MessageConverter`的引用，用于在 JMS 消息与 Spring Integration 消息有效载荷（或如果 `extract-request-payload`为 `false`则为消息）之间进行转换。默认值是一个 `SimpleMessageConverter。
8	请求消息的默认优先级。如果存在，则会被消息优先级标头覆盖。它的范围是 `0` 到 `9`。此设置仅在 `explicit-qos-enabled`为 `true`时生效。
9	等待回复的时间（以毫秒为单位）。默认值为 `5000`（五秒）。
10	回复消息发送到的通道。
11	一个 Destination`的引用，它被设置为 `JMSReplyTo`标头。最多只能使用一个 `reply-destination、reply-destination-expression`或 `reply-destination-name。如果没有提供任何一个，则使用 `TemporaryQueue`来回复此网关。
12	一个 SpEL 表达式，计算为一个 `Destination`，它将被设置为 JMSReplyTo`标头。该表达式的结果可以是 `Destination`对象或 `String。它由 DestinationResolver`用来解析实际的 `Destination。最多只能使用一个 `reply-destination`、reply-destination-expression`或 `reply-destination-name。如果没有提供任何一个，则使用 `TemporaryQueue`来回复此网关。
13	设定为 JMSReplyTo 标头的目标名称。它由 `DestinationResolver` 用来解析实际的 `Destination`。最多只允许一个 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name`。如果没有提供，则将使用 `TemporaryQueue` 来回复此网关。
14	当设定为 `true` 时，它表示任何由 `DestinationResolver` 解析出的回复 `Destination` 应当是 `Topic` 而不是 `Queue`。
15	网关在将回复消息发送到 `reply-channel` 时等待的时间。这只有当 `reply-channel` 能够阻塞时才有效果 — 例如一个容量有限并且当前已满的 `QueueChannel`。默认值是无限大。
16	此网关接收请求消息的通道。
17	对接收请求消息的 `Destination` 的引用。必须有一个 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name`。你只能使用这三个属性中的一个。
18	一个求值为 `Destination` 的 SpEL 表达式，请求消息被发送到该 `Destination`。该表达式可以产生一个 `Destination` 对象或一个 `String`。它由 `DestinationResolver` 用来解析实际的 `Destination`。必须有一个 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name`。你只能使用这三个属性中的一个。
19	发送请求消息到的目标的名称。它由 `DestinationResolver` 用来解析实际的 `Destination`。必须有一个 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name`。你只能使用这三个属性中的一个。
20	当设定为 `true` 时，它表示任何由 `DestinationResolver` 解析出的请求 `Destination` 应当是 `Topic` 而不是 `Queue`。
21	指定消息生存时间。只有当 `explicit-qos-enabled` 为 `true` 时此设定才会生效。
22	指定此出站网关是否必须返回一个非空值。默认情况下，此值为 `true`，并且当底层服务在 `receive-timeout` 之后不返回一个值时，将抛出一个 `MessageTimeoutException`。请注意，如果服务根本不期望返回一个回复，那么最好使用一个具有 `requires-reply="false"` 的 `<int-jms:outbound-channel-adapter/>`，而不是使用一个 `<int-jms:outbound-gateway/>`。对于后者，发送线程会被阻塞，等待一个 `receive-timeout` 周期的回复。
23	当你使用一个 `<reply-listener />` 时，其生命周期（启动和停止）默认情况下与网关的生命周期匹配。当此值大于 `0` 时，容器会根据需要启动（在发送请求时）。容器会持续运行，直到在此期间至少未收到任何请求（且没有回复悬而未决）。容器会在下一次请求时再次启动。停止时间是一个最小值，实际时长可能长达此值的 1.5 倍。
24	See Async Gateway.
25	当包含这个元素时，回复将由一个异步 `MessageListenerContainer` 接收，而不是为每个回复创建一个消费者。在许多情况下，这将更有效率。

== Mapping Message Headers to and from JMS Message

JMS 消息可以包含元信息，例如 JMS API 头和简单属性。你可以使用 JmsHeaderMapper 将它们映射到 Spring Integration 消息头和从 Spring Integration 消息头映射。JMS API 头传递给适当的 setter 方法（例如 setJMSReplyTo），而其他头则复制到 JMS 消息的一般属性。JMS 出站网关使用 JmsHeaderMapper 的默认实现自举，该实现将映射标准 JMS API 头以及原语或 String 消息头。你还可以使用入站和出站网关的 header-mapper 属性提供自定义头映射器。

许多 JMS 厂商特定的客户端不允许在已创建的 JMS 消息上直接设置 deliveryMode、priority 和 redeliveryDelay 属性。它们被认为是 QoS 属性，因此必须传播到目标 JMS API。出于此原因，Outbound Channel Adapter 不将相应的 Spring Integration 头（或表达式结果）映射到所提到的 JMS 消息属性。相反，Outbound Channel Adapter 使用 MessageBuilder，将请求消息中的头值传播到 JMS API。要启用此功能，必须使用 deliveryMode 属性为 true 配置 outbound 端点。Spring Integration Java DSL 默认为 outbound 端点配置 deliveryMode，但仍然必须设置 delay 属性。

从 4.0 版本开始，JMSPriority 标头已映射到标准 priority 标头，用于入站消息。先前，priority 标头仅用于出站消息。若要还原到之前的行为（即，不映射入站优先级），请将 DefaultJmsHeaderMapper 的 mapInboundPriority 属性设置为 false。

从 4.3 版本开始，DefaultJmsHeaderMapper 通过调用其 toString() 方法将标准 correlationId 标头映射为消息属性（correlationId 通常是 UUID，不被 JMS 支持）。在入站方面，它被映射为 String。这与 jms_correlationId 标头无关，后者映射至 JMSCorrelationID 标头并反向映射。JMSCorrelationID 通常用于关联请求和答复，而 correlationId 通常用于将相关消息合并到一个组中（例如，与聚合器或重新排序器一起）。

从版本 5.1 开始，可以配置 DefaultJmsHeaderMapper 来映射入站 JMSDeliveryMode 和 JMSExpiration 属性：

@Bean
public DefaultJmsHeaderMapper jmsHeaderMapper() {
    DefaultJmsHeaderMapper mapper = new DefaultJmsHeaderMapper();
    mapper.setMapInboundDeliveryMode(true)
    mapper.setMapInboundExpiration(true)
    return mapper;
}

这些 JMS 属性分别映射到 Spring 消息头 JmsHeaders.DELIVERY_MODE 和 JmsHeaders.EXPIRATION。

== Message Conversion, Marshalling, and Unmarshalling

如果你需要转换消息，则所有 JMS 适配器和网关都允许你通过设置 message-converter 属性来提供 MessageConverter 。为此，请提供同一 ApplicationContext 中可用的 MessageConverter 实例的 bean 名称。此外，为了与 marshaller 和 unmarshaller 接口保持一致，Spring 提供了 MarshallingMessageConverter，你可以使用你自己的自定义 marshaller 和 unmarshaller 对其进行配置。以下示例展示了如何执行此操作：

<int-jms:inbound-gateway request-destination="requestQueue"
    request-channel="inbound-gateway-channel"
    message-converter="marshallingMessageConverter"/>

<bean id="marshallingMessageConverter"
    class="org.springframework.jms.support.converter.MarshallingMessageConverter">
    <constructor-arg>
        <bean class="org.bar.SampleMarshaller"/>
    </constructor-arg>
    <constructor-arg>
        <bean class="org.bar.SampleUnmarshaller"/>
    </constructor-arg>
</bean>

当您提供自己的 MessageConverter 实例时，它仍会包含在 HeaderMappingMessageConverter 中。这意味着“提取请求有效负载”和“提取答复有效负载”属性会影响传递给转换器的实际对象。HeaderMappingMessageConverter 本身委托给目标 MessageConverter，同时还将 Spring Integration MessageHeaders 映射到 JMS 消息属性，然后反向映射。

== JMS-backed Message Channels

前面介绍的通道适配器和网关都适用于与其他外部系统集成的应用程序。入站选项假定某些其他系统正在向 JMS 目标发送 JMS 消息，并且出站选项假定某些其他系统正在从该目标接收。其他系统可能是或可能不是 Spring Integration 应用程序。当然，当将 Spring Integration 消息实例本身作为 JMS 消息主体发送时（将 'extract-payload' 值设置为 false），则假定其他系统基于 Spring Integration。然而，这绝不是必须的。这种灵活性是使用带有“channels”（或 JMS 中的目标）抽象的消息传递集成选项的优点之一。

有时，给定 JMS 目标的生产者和消费者都旨在成为同一应用程序的一部分，在同一个进程中运行。你可以使用一对入站和出站通道适配器来完成此操作。这种方法的问题在于你需要两个适配器，即使从概念上来说，目标是拥有一个消息通道。从 Spring Integration 版本 2.0 开始支持更好的选择。现在在使用 JMS 命名空间时可以定义一个 "`channel“` ，如下例所示：

<int-jms:channel id="jmsChannel" queue="exampleQueue"/>

上一个示例中的通道的行为与主 Spring Integration 命名空间中的普通 <channel/> 元素非常相似。它可以被任何端点的 input-channel 和 output-channel 属性引用。不同之处在于此通道由名为 exampleQueue 的 JMS 队列实例支持。这意味着可以在生产端点和消费端点之间进行异步消息传递。但是，与通过在非 JMS <channel/> 元素中添加 <queue/> 元素创建的更简单的异步消息通道不同，这些消息不会存储在内存队列中。相反，这些消息在 JMS 消息主体中传递，然后可用于该通道的底层 JMS 提供程序的全部功能。使用此替代方案的最常见理由可能是利用 JMS 消息传递的存储转发方法提供的持久性。

如果配置正确，JMS 支持的消息通道还支持事务。换句话说，如果生产者的发送操作是回滚事务的一部分，那么生产者实际上不会向事务性 JMS 支持的通道写入。同样地，如果对该消息的接收是回滚事务的一部分，则消费者不会从通道中物理删除 JMS 消息。请注意，在此方案中，生产者和消费者事务是分开的。这与传播事务上下文到没有 <queue/> 子元素的简单同步 <channel/> 元素有很大不同。

由于上一个示例引用了一个 JMS 队列实例，因此它充当一个点对点通道。另一方面，如果你需要发布订阅行为，则可以使用一个单独的元素并引用一个 JMS 主题。以下示例展示了如何执行此操作：

<int-jms:publish-subscribe-channel id="jmsChannel" topic="exampleTopic"/>

对于两种类型的 JMS 支持的通道，都可以提供目标名称而不是引用，如下例所示：

<int-jms:channel id="jmsQueueChannel" queue-name="exampleQueueName"/>

<jms:publish-subscribe-channel id="jmsTopicChannel" topic-name="exampleTopicName"/>

在以上示例中，目标名称由 Spring 的默认 DynamicDestinationResolver 实现解析，但你可以提供 DestinationResolver 接口的任何实现。另外，JMS ConnectionFactory 是通道的必备属性，但默认情况下，预期的 bean 名称将为 jmsConnectionFactory。以下示例提供了用于解析 JMS 目标名称的自定义实例和 ConnectionFactory 的不同名称：

<int-jms:channel id="jmsChannel" queue-name="exampleQueueName"
    destination-resolver="customDestinationResolver"
    connection-factory="customConnectionFactory"/>

对于 <publish-subscribe-channel />，将 durable 属性设置为 true 以启用持久订阅，或将该属性设置为 subscription-shared 以启用共享订阅（需要 JMS 2.0 代理，且版本 4.2 之后一直可用）。使用 subscription 来命名订阅。

== Using JMS Message Selectors

使用 JMS 消息选择器，您既可以根据 JMS 标头也可以根据 JMS 属性来筛选 JMS Messages。例如，如果您想收听自定义 JMS 标头属性为 myHeaderProperty、等于 something 的消息，您可以指定以下表达式：

myHeaderProperty = 'something'

消息选择器表达式是 SQL-92 条件表达式语法的子集，并且被定义为 Java Message Service 规范的一部分。您可以使用 XML 命名空间配置为以下 Spring Integration JMS 组件指定 JMS 消息 selector 属性：

JMS Channel
JMS Publish Subscribe Channel
JMS Inbound Channel Adapter
JMS Inbound Gateway
JMS Message-driven Channel Adapter

您无法使用 JMS 消息选择器来引用消息体值。

== JMS Samples

要尝试使用这些 JMS 适配器，请查看 Spring Integration Samples Git 存储库中提供的 JMS 样品，网址为 https://github.com/spring-projects/spring-integration-samples/tree/master/basic/jms.

该存储库包括两个样本。一个提供入站和出站通道适配器，另一个提供入站和出站网关。它们被配置为使用嵌入式 ActiveMQ 进程运行，但是您可以修改每个样本的 common.xml Spring 应用程序上下文文件，以支持不同的 JMS 提供程序或独立的 ActiveMQ 进程。

换句话说，你可以拆分配置，以便入站和出站适配器在单独的 JVM 中运行。如果你已安装 ActiveMQ，请将 common.xml 文件中的 brokerURL 属性修改为使用 tcp://localhost:61616（而不是 vm://localhost）。这两个示例都接受来自标准输入的输入，并回显到标准输出。查看配置，了解如何通过 JMS 路由这些消息。

JMS Support

Inbound Channel Adapter

Transactions

Message Driven Channel Adapter

Inbound Conversion Errors

Outbound Channel Adapter

Transactions

Inbound Gateway

Outbound Gateway

Using a <reply-listener/>

Idle Reply Listeners

Gateway Reply Correlation

Async Gateway

Attribute Reference

Using a `<reply-listener/>`