JMS 支持

Spring Integration 提供通道适配器，用于接收和发送 JMS 消息。

您需要将此依赖项包含到您的项目中

  • Maven

  • Gradle

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-jms</artifactId>
    <version>6.3.0</version>
</dependency>
compile "org.springframework.integration:spring-integration-jms:6.3.0"

jakarta.jms:jakarta.jms-api 必须通过一些 JMS 供应商特定的实现显式添加，例如 Apache ActiveMQ。

实际上存在两种基于 JMS 的入站通道适配器。第一个使用 Spring 的 JmsTemplate 基于轮询周期接收消息。第二个是“消息驱动”的，依赖于 Spring 的 MessageListener 容器。出站通道适配器使用 JmsTemplate 按需转换和发送 JMS 消息。

通过使用 JmsTemplate 和 MessageListener 容器，Spring Integration 依赖于 Spring 的 JMS 支持。这很重要，因为这些适配器上公开的大多数属性都配置了底层的 JmsTemplate 和 MessageListener 容器。有关 JmsTemplate 和 MessageListener 容器的更多详细信息，请参阅 Spring JMS 文档。

虽然 JMS 通道适配器旨在用于单向消息传递（仅发送或仅接收），但 Spring Integration 还提供入站和出站 JMS 网关用于请求和回复操作。入站网关依赖于 Spring 的 MessageListener 容器实现之一来进行消息驱动的接收。它还能够将返回值发送到接收消息提供的 reply-to 目的地。出站网关将 JMS 消息发送到 request-destination（或 request-destination-name 或 request-destination-expression），然后接收回复消息。您可以显式配置 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 供应商对这些目的地的处理方式不同。

入站通道适配器

入站通道适配器需要引用单个 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>
从前面的配置中可以注意到，inbound-channel-adapter 是一个轮询消费者。这意味着它在触发时会调用 receive()。您应该只在轮询相对不频繁且时间性不重要的场景中使用它。对于所有其他场景（绝大多数基于 JMS 的用例），message-driven-channel-adapter (稍后描述) 是更好的选择。
默认情况下，所有需要引用 ConnectionFactory 的 JMS 适配器都会自动查找名为 jmsConnectionFactory 的 Bean。这就是为什么在许多示例中您没有看到 connection-factory 属性的原因。但是，如果您的 JMS ConnectionFactory 有不同的 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 版本开始，org.springframework.jms.connection.CachingConnectionFactory 和 cacheConsumers 的 receive-timeout 默认值为 -1（不等待），否则为 1 秒。JMS 入站通道适配器基于提供的 ConnectionFactory 和选项创建 DynamicJmsTemplate。如果需要外部 JmsTemplate（例如在 Spring Boot 环境中），或者 ConnectionFactory 不进行缓存，或者没有 cacheConsumers，建议设置 jmsTemplate.receiveTimeout(-1)，如果预期非阻塞消费。

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

事务

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

但是，请注意，将 session-transacted 设置为 true 意义不大，因为事务会在 receive() 操作之后，消息发送到 channel 之前立即提交。

如果您希望整个流程都处于事务中（例如，如果存在下游出站通道适配器），则必须使用带有 JmsTransactionManager 的 transactional 轮询器。或者，考虑使用 jms-message-driven-channel-adapter，并将 acknowledge 设置为 transacted（默认值）。

消息驱动通道适配器

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 版本开始，默认的 acknowledge 模式为 transacted，除非您提供外部容器。在这种情况下，您应该根据需要配置容器。我们建议在 DefaultMessageListenerContainer 中使用 transacted，以避免消息丢失。

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

最后，<message-driven-channel-adapter> 元素也接受 'error-channel' 属性。这提供了与进入 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<?>> 用于下游发送和发送-接收操作的重试。这些选项也暴露在 JmsMessageDrivenChannelAdapterSpec 中，用于 Java DSL。

入站转换错误

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

出站通道适配器

JmsSendingMessageHandler 实现 MessageHandler 接口，能够将 Spring Integration Messages 转换为 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。另一方面，如果您想将实际的 Spring Integration 消息通过 JMS 发送到另一个系统，请将其设置为“false”。

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

从 5.1 版本开始，<int-jms:outbound-channel-adapter>（JmsSendingMessageHandler）可以使用 deliveryModeExpression 和 timeToLiveExpression 属性进行配置，以根据请求 Spring Message 在运行时评估 JMS 消息发送的适当 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()"/>

事务

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

入站网关

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

由于网关提供请求-回复行为而不是单向发送或接收行为，因此它们也具有两个不同的属性用于“有效负载提取”（如 前面讨论的 通道适配器的 'extract-payload' 设置）。对于入站网关，'extract-request-payload' 属性决定是否提取收到的 JMS 消息体。如果为 'false'，则 JMS 消息本身成为 Spring Integration 消息有效负载。默认值为 'true'。

类似地，对于入站网关，'extract-reply-payload' 属性适用于要转换为回复 JMS 消息的 Spring Integration 消息。如果您想传递整个 Spring Integration 消息（作为 JMS ObjectMessage 的主体），请将此值设置为 'false'。默认情况下，它也是 'true'，即 Spring Integration 消息有效负载将转换为 JMS 消息（例如，String 有效负载将变为 JMS TextMessage）。

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

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

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

您可能会注意到，此示例与 进入 GatewayProxyFactoryBean 中包含的示例非常相似。这里也适用相同的理念：exceptionTransformer 可以是创建错误响应对象的 POJO，您可以引用 nullChannel 来抑制错误，或者您可以省略 'error-channel' 以让异常传播。

请参阅 入站转换错误。

从主题消费时，将 pub-sub-domain 属性设置为 true。将 subscription-durable 设置为 true 以创建持久订阅，或将 subscription-shared 设置为 true 以创建共享订阅（需要 JMS 2.0 代理，自版本 4.2 起可用）。使用 subscription-name 为订阅命名。

从版本 4.2 开始，默认的 acknowledge 模式为 transacted，除非提供外部容器。在这种情况下，您应该根据需要配置容器。我们建议您将 transacted 与 DefaultMessageListenerContainer 一起使用，以避免消息丢失。

从版本 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() 选项，用于重试内部发送和接收操作。

出站网关

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

将reply-destination（或reply-destination-name）与cacheConsumers设置为true的CachingConnectionFactory一起使用会导致内存不足的情况。这是因为每个请求都会获得一个新的消费者，该消费者具有一个新的选择器（选择correlation-key值，或者当没有correlation-key时，选择发送的JMSMessageID）。鉴于这些选择器是唯一的，它们在当前请求完成之后会保留在缓存中（未被使用）。

如果您指定了回复目的地，建议您不要使用缓存的消费者。或者，可以考虑使用<reply-listener/>，如以下所述。

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

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

“出站网关”有效载荷提取属性与“入站网关”的属性相反（参见之前的讨论）。这意味着“extract-request-payload”属性值适用于将Spring Integration消息转换为JMS消息以作为请求发送。而“extract-reply-payload”属性值适用于作为回复接收到的JMS消息，然后将其转换为Spring Integration消息，随后发送到“reply-channel”，如前面的配置示例所示。

使用<reply-listener/>

Spring Integration 2.2 引入了一种处理回复的替代技术。如果您向网关添加<reply-listener/>子元素，而不是为每个回复创建消费者，则会使用MessageListener容器来接收回复并将它们传递给请求线程。这提供了许多性能优势，并缓解了之前注意事项中描述的缓存消费者内存利用问题。

当使用<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 文档。

空闲回复监听器

从 4.2 版本开始，您可以根据需要启动回复监听器（并在空闲一段时间后停止它），而不是在网关的生命周期内运行。如果您在应用程序上下文中拥有许多网关，而它们大多处于空闲状态，这将非常有用。一种这样的情况是，上下文包含许多（非活动）分区Spring Batch 作业，这些作业使用 Spring Integration 和 JMS 进行分区分配。如果所有回复监听器都处于活动状态，则 JMS 代理会为每个网关创建一个活动消费者。通过启用空闲超时，每个消费者仅在相应的批处理作业运行时（以及作业完成后的短时间内）存在。

请参见属性参考中的idle-reply-listener-timeout。

网关回复关联

本节描述用于回复关联（确保原始网关仅接收对其请求的回复）的机制，具体取决于网关的配置方式。有关此处讨论的属性的完整描述，请参见属性参考。

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

  1. 没有reply-destination*属性，也没有<reply-listener>

    为每个请求创建一个TemporaryQueue，并在请求完成（成功或失败）时删除。correlation-key无关紧要。

  2. 提供了一个reply-destination*属性，但没有提供<reply-listener/>或correlation-key

    将等于发送消息的 JMSCorrelationID 用作消费者的消息选择器

    messageSelector = "JMSCorrelationID = '" + messageId + "'"

    响应系统应在回复 JMSCorrelationID 中返回传入的 JMSMessageID。这是一种常见模式，由 Spring Integration 入站网关以及 Spring 的 MessageListenerAdapter 用于消息驱动的 POJO 实现。

    使用此配置时，不应使用主题进行回复。回复可能会丢失。

  3. 提供 reply-destination* 属性，不提供 <reply-listener/>，并且 correlation-key="JMSCorrelationID"

    网关生成一个唯一的关联 ID 并将其插入 JMSCorrelationID 标头。消息选择器是

    messageSelector = "JMSCorrelationID = '" + uniqueId + "'"

    响应系统应在回复 JMSCorrelationID 中返回传入的 JMSCorrelationID。这是一种常见模式，由 Spring Integration 入站网关以及 Spring 的 MessageListenerAdapter 用于消息驱动的 POJO 实现。

  4. 提供 reply-destination* 属性，不提供 <reply-listener/>，并且 correlation-key="myCorrelationHeader"

    网关生成一个唯一的关联 ID 并将其插入 myCorrelationHeader 消息属性。correlation-key 可以是任何用户定义的值。消息选择器是

    messageSelector = "myCorrelationHeader = '" + uniqueId + "'"

    响应系统应在回复 myCorrelationHeader 中返回传入的 myCorrelationHeader。

  5. 提供 reply-destination* 属性，不提供 <reply-listener/>，并且 correlation-key="JMSCorrelationID*"（注意关联键中的 *）。

    网关使用请求消息中 jms_correlationId 标头（如果存在）中的值，并将其插入 JMSCorrelationID 标头。消息选择器是

    messageSelector = "JMSCorrelationID = '" + headers['jms_correlationId'] + "'"

    用户必须确保此值是唯一的。

    如果标头不存在，网关的行为与 3 中相同。

    响应系统应在回复 JMSCorrelationID 中返回传入的 JMSCorrelationID。这是一种常见模式，由 Spring Integration 入站网关以及 Spring 的 MessageListenerAdapter 用于消息驱动的 POJO 实现。

  6. 不提供 reply-destination* 属性，并提供 <reply-listener>

    将创建一个临时队列，并将其用于来自此网关实例的所有回复。消息中不需要关联数据，但网关内部使用发送的 JMSMessageID 将回复定向到正确的请求线程。

  7. 提供 reply-destination* 属性，提供 <reply-listener>，并且不提供 correlation-key

    不允许。

    <reply-listener/> 配置将被忽略，网关的行为与 2 中相同。将写入警告日志消息以指示这种情况。

  8. 提供 reply-destination* 属性，提供 <reply-listener>，并且 correlation-key="JMSCorrelationID"

    网关具有一个唯一的关联 ID，并将其与 JMSCorrelationID 标头中的递增值一起插入（gatewayId + "_" + ++seq）。消息选择器是

    messageSelector = "JMSCorrelationID LIKE '" + gatewayId%'"

    响应系统应在回复的 JMSCorrelationID 中返回传入的 JMSCorrelationID。这是一种常见的模式，由 Spring Integration 入站网关以及 Spring 的 MessageListenerAdapter 用于消息驱动的 POJO 实现。由于每个网关都有一个唯一的 ID，因此每个实例只接收自己的回复。完整的关联数据用于将回复路由到正确的请求线程。

  9. 提供 reply-destination* 属性，提供 <reply-listener/>，并提供 correlation-key="myCorrelationHeader"

    网关具有唯一的关联 ID，并将其与 myCorrelationHeader 属性中的递增值一起插入（gatewayId + "_" + ++seq）。correlation-key 可以是任何用户定义的值。消息选择器是

    messageSelector = "myCorrelationHeader LIKE '" + gatewayId%'"

    响应系统应在回复的 myCorrelationHeader 中返回传入的 myCorrelationHeader。由于每个网关都有一个唯一的 ID，因此每个实例只接收自己的回复。完整的关联数据用于将回复路由到正确的请求线程。

  10. 提供 reply-destination* 属性，提供 <reply-listener/>，并提供 correlation-key="JMSCorrelationID*"

    （注意关联键中的 *）

    不允许。

    使用回复侦听器不允许使用用户提供的关联 ID。网关不会使用此配置进行初始化。

异步网关

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

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

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

属性参考

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

将消息头映射到 JMS 消息和从 JMS 消息映射消息头

JMS 消息可以包含元信息，例如 JMS API 头部和简单属性。您可以使用 `JmsHeaderMapper` 将这些信息映射到 Spring Integration 消息头部，反之亦然。JMS API 头部将传递给相应的 setter 方法（例如 `setJMSReplyTo`），而其他头部将复制到 JMS 消息的通用属性中。JMS 出站网关使用 `JmsHeaderMapper` 的默认实现进行引导，该实现将映射标准 JMS API 头部以及原始或 `String` 消息头部。您也可以使用入站和出站网关的 `header-mapper` 属性提供自定义头部映射器。

许多 JMS 供应商特定的客户端不允许直接在已创建的 JMS 消息上设置 `deliveryMode`、`priority` 和 `timeToLive` 属性。它们被认为是 QoS 属性，因此必须传播到目标 `MessageProducer.send(message, deliveryMode, priority, timeToLive)` API。因此，`DefaultJmsHeaderMapper` 不会将相应的 Spring Integration 头部（或表达式结果）映射到上述 JMS 消息属性中。相反，`JmsSendingMessageHandler` 使用 `DynamicJmsTemplate` 将请求消息中的头部值传播到 `MessageProducer.send()` API。要启用此功能，您必须使用 `explicitQosEnabled` 属性设置为 true 的 `DynamicJmsTemplate` 配置出站端点。Spring Integration Java DSL 默认配置 `DynamicJmsTemplate`，但您仍然需要设置 `explicitQosEnabled` 属性。
从 4.0 版本开始，`JMSPriority` 头部映射到入站消息的标准 `priority` 头部。以前，`priority` 头部仅用于出站消息。要恢复到之前的行为（即不映射入站优先级），请将 `DefaultJmsHeaderMapper` 的 `mapInboundPriority` 属性设置为 `false`。
从 4.3 版本开始，DefaultJmsHeaderMapper 通过调用其 toString() 方法将标准 correlationId 标头映射为消息属性（correlationId 通常是 UUID，JMS 不支持）。在入站侧，它被映射为 String。这与 jms_correlationId 标头无关，jms_correlationId 标头映射到 JMSCorrelationID 标头并从 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 属性分别映射到 JmsHeaders.DELIVERY_MODE 和 JmsHeaders.EXPIRATION Spring 消息标头。

消息转换、编组和反编组

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

<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 中。这意味着“extract-request-payload”和“extract-reply-payload”属性会影响传递给转换器的实际对象。HeaderMappingMessageConverter 本身会委托给目标 MessageConverter，同时也将 Spring Integration MessageHeaders 映射到 JMS 消息属性并反之。

基于 JMS 的消息通道

前面介绍的通道适配器和网关都旨在用于与其他外部系统集成的应用程序。入站选项假设其他系统正在将 JMS 消息发送到 JMS 目的地，而出站选项假设其他系统正在从目的地接收消息。另一个系统可能是也可能不是 Spring Integration 应用程序。当然，当将 Spring Integration 消息实例作为 JMS 消息本身的主体发送时（将“extract-payload”值设置为false），则假设另一个系统基于 Spring Integration。但是，这绝不是必需的。这种灵活性是使用基于消息的集成选项与“通道”（或 JMS 情况下的目的地）抽象带来的优势之一。

有时，给定 JMS 目地的生产者和消费者都打算成为同一应用程序的一部分，在同一进程中运行。您可以通过使用一对入站和出站通道适配器来实现这一点。这种方法的问题在于，即使从概念上讲，目标是拥有一个消息通道，您也需要两个适配器。从 Spring Integration 2.0 版本开始，支持更好的选项。现在，使用 JMS 命名空间时，可以定义一个“通道”，如下面的示例所示

<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 消息。请注意，在这种情况下，生产者和消费者事务是分开的。这与通过简单的同步 <channel/> 元素（没有 <queue/> 子元素）传播事务上下文有很大不同。

由于上面的示例引用了 JMS Queue 实例，因此它充当点对点通道。另一方面，如果您需要发布订阅行为，可以使用单独的元素并引用 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 来命名订阅。

使用 JMS 消息选择器

使用 JMS 消息选择器，您可以根据 JMS 标头和 JMS 属性过滤 JMS 消息。例如，如果您想监听自定义 JMS 标头属性 myHeaderProperty 等于 something 的消息，您可以指定以下表达式。

myHeaderProperty = 'something'

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

  • JMS 通道

  • JMS 发布订阅通道

  • JMS 入站通道适配器

  • JMS 入站网关

  • JMS 消息驱动通道适配器

您不能使用 JMS 消息选择器引用消息正文值。

JMS 示例

要尝试使用这些 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://127.0.0.1:61616（而不是 vm://127.0.0.1）。这两个示例都接受来自 stdin 的输入并回显到 stdout。查看配置以了解这些消息是如何通过 JMS 传输的。