JMS 支持
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}"
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>
请注意,从前面的配置中可以看出, |
默认情况下,所有需要引用 |
如果 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))
事务
从 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"/>
消息驱动适配器还接受与 |
你不能使用 Spring JMS 命名空间元素 |
从 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
设置为 true
以进行共享订阅(需要 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")))
.get();
}
@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”。
无论有效负载提取的布尔值如何,只要你依赖默认转换器或提供对 |
从 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()"/>
入站网关
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”。
默认情况下,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
,除非提供外部容器。
在这种情况下,你应该根据需要配置容器。
我们建议将 DefaultMessageListenerContainer
与 transacted
一起使用,以避免消息丢失。
从 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
接收,或者,如果没有提供 reply-destination
,则通过创建 JMS TemporaryQueue
(如果 replyPubSubDomain= true
,则创建 TemporaryTopic
)实例来处理 JMS 回复消息。
将 reply-destination
(或 reply-destination-name
)与 CachingConnectionFactory
结合使用,如果 cacheConsumers
设置为 true
,可能会导致内存不足。
这是因为每个请求都会获得一个带有新选择器的新消费者(根据 correlation-key
值或在没有 correlation-key
时根据发送的 JMSMessageID 进行选择)。
鉴于这些选择器是唯一的,它们在当前请求完成后仍保留在缓存中(未使用)。
如果指定回复目的地,建议不要使用缓存的消费者。
或者,考虑使用 jms-outbound-gateway-reply-listener <reply-listener/>
。
以下示例显示了如何配置出站网关:
<int-jms:outbound-gateway id="jmsOutGateway"
request-destination="outQueue"
request-channel="outboundJmsRequests"
reply-channel="jmsReplies"/>
“出站网关”的有效负载提取属性与“入站网关”的属性相反(请参阅 前面的讨论)。 这意味着“extract-request-payload”属性值适用于转换为 JMS 消息并作为请求发送的 Spring Integration 消息。 “extract-reply-payload”属性值适用于作为回复接收的 JMS 消息,然后转换为 Spring Integration 消息,随后发送到“reply-channel”,如前面的配置示例所示。
使用 <reply-listener/>
Spring Integration 2.2 引入了一种处理回复的替代技术。
如果将 <reply-listener/>
子元素添加到网关而不是为每个回复创建消费者,则使用 MessageListener
容器接收回复并将其交给请求线程。
这提供了许多性能优势,并缓解了 jms-outbound-gateway-memory-caution 缓存消费者内存利用率问题。
当使用带有没有 reply-destination
的出站网关的 <reply-listener/>
时,不是为每个请求创建一个 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 代理为每个网关都有一个活动的消费者。 通过启用空闲超时,每个消费者仅在其相应的批处理作业运行时(以及作业完成后的短时间内)存在。
请参阅 jms-og-attributes 中的 idle-reply-listener-timeout
。
网关回复关联
本节描述了用于回复关联的机制(确保源网关只接收对其请求的回复),具体取决于网关的配置方式。 有关此处讨论的属性的完整描述,请参阅 jms-og-attributes。
以下列表描述了各种场景(数字用于标识——顺序无关紧要):
-
没有
reply-destination*
属性,也没有<reply-listener>
。 为每个请求创建一个TemporaryQueue
,并在请求完成时(成功或失败)删除。correlation-key
无关紧要。 -
提供了
reply-destination*
属性,但未提供<reply-listener/>
和correlation-key
。JMSCorrelationID
等于出站消息的值用作消费者的消息选择器:messageSelector = "JMSCorrelationID = '" + messageId + "'"
响应系统应在回复JMSCorrelationID
中返回入站JMSMessageID
。 这是一种常见的模式,由 Spring Integration 入站网关以及 Spring 的MessageListenerAdapter
用于消息驱动 POJO 实现。
使用此配置时,不应将主题用于回复。 回复可能会丢失。 |
-
提供了
reply-destination*
属性,未提供<reply-listener/>
,且correlation-key="JMSCorrelationID"
。 网关生成一个唯一的关联 ID,并将其插入JMSCorrelationID
头部。 消息选择器为:messageSelector = "JMSCorrelationID = '" + uniqueId + "'"
响应系统应在回复JMSCorrelationID
中返回入站JMSCorrelationID
。 这是一种常见的模式,由 Spring Integration 入站网关以及 Spring 的MessageListenerAdapter
用于消息驱动 POJO 实现。 -
提供了
reply-destination*
属性,未提供<reply-listener/>
,且correlation-key="myCorrelationHeader"
。 网关生成一个唯一的关联 ID,并将其插入myCorrelationHeader
消息属性中。correlation-key
可以是任何用户定义的值。 消息选择器为:messageSelector = "myCorrelationHeader = '" + uniqueId + "'"
响应系统应在回复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-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" [id="CO1-1"]1
correlation-key="" [id="CO1-2"]2
delivery-persistent="" [id="CO1-3"]3
destination-resolver="" [id="CO1-4"]4
explicit-qos-enabled="" [id="CO1-5"]5
extract-reply-payload="true" [id="CO1-6"]6
extract-request-payload="true" [id="CO1-7"]7
header-mapper="" [id="CO1-8"]8
message-converter="" [id="CO1-9"]9
priority="" [id="CO1-10"]10
receive-timeout="" [id="CO1-11"]11
reply-channel="" [id="CO1-12"]12
reply-destination="" [id="CO1-13"]13
reply-destination-expression="" [id="CO1-14"]14
reply-destination-name="" [id="CO1-15"]15
reply-pub-sub-domain="" [id="CO1-16"]16
reply-timeout="" [id="CO1-17"]17
request-channel="" [id="CO1-18"]18
request-destination="" [id="CO1-19"]19
request-destination-expression="" [id="CO1-20"]20
request-destination-name="" [id="CO1-21"]21
request-pub-sub-domain="" [id="CO1-22"]22
time-to-live="" [id="CO1-23"]23
requires-reply="" [id="CO1-24"]24
idle-reply-listener-timeout="" [id="CO1-25"]25
async=""> [id="CO1-26"]26
<int-jms:reply-listener /> [id="CO1-27"]27
</int-jms:outbound-gateway>
<1> 对 `jakarta.jms.ConnectionFactory` 的引用。 默认值为 `jmsConnectionFactory`。 <1> 包含关联数据的属性名称,用于关联响应与回复。 如果省略,网关期望响应系统在 `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` 时生效。 <1> 一个 `DestinationResolver`。 默认是一个 `DynamicDestinationResolver`,它将目的地名称映射到具有该名称的队列或主题。 <1> 当设置为 `true` 时,它启用质量服务属性的使用:`priority`、`delivery-mode` 和 `time-to-live`。 <1> 当设置为 `true`(默认值)时,Spring Integration 回复消息的有效负载是从 JMS 回复消息的主体(使用 `MessageConverter`)创建的。 当设置为 `false` 时,整个 JMS 消息成为 Spring Integration 消息的有效负载。 <1> 当设置为 `true`(默认值)时,Spring Integration 消息的有效负载将转换为 `JMSMessage`(使用 `MessageConverter`)。 当设置为 `false` 时,整个 Spring Integration 消息将转换为 `JMSMessage`。 在这两种情况下,Spring Integration 消息头都通过 `HeaderMapper` 映射到 JMS 头和属性。 <1> 一个 `HeaderMapper`,用于将 Spring Integration 消息头映射到 JMS 消息头和属性之间。 <1> 对 `MessageConverter` 的引用,用于在 JMS 消息和 Spring Integration 消息有效负载(或消息,如果 `extract-request-payload` 为 `false`)之间进行转换。 默认是一个 `SimpleMessageConverter`。 <1> 请求消息的默认优先级。 如果存在消息优先级头,则会被覆盖。 其范围是 `0` 到 `9`。 此设置仅在 `explicit-qos-enabled` 为 `true` 时生效。 <1> 等待回复的时间(以毫秒为单位)。 默认值为 `5000`(五秒)。 <1> 回复消息发送到的通道。 <1> 对 `Destination` 的引用,该 `Destination` 设置为 `JMSReplyTo` 头部。 最多只允许 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name` 中的一个。 如果未提供,则使用 `TemporaryQueue` 作为此网关的回复。 <1> 一个 SpEL 表达式,评估为一个 `Destination`,它将设置为 `JMSReplyTo` 头部。 表达式可以生成 `Destination` 对象或 `String`。 它由 `DestinationResolver` 用于解析实际的 `Destination`。 最多只允许 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name` 中的一个。 如果未提供,则使用 `TemporaryQueue` 作为此网关的回复。 <1> 设置为 JMSReplyTo 头的目标名称。 它由 `DestinationResolver` 用于解析实际的 `Destination`。 最多只允许 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name` 中的一个。 如果未提供,则使用 `TemporaryQueue` 作为此网关的回复。 <1> 当设置为 `true` 时,表示由 `DestinationResolver` 解析的任何回复 `Destination` 都应该是 `Topic` 而不是 `Queue`。 <1> 网关在发送回复消息到 `reply-channel` 时等待的时间。 这仅在 `reply-channel` 可以阻塞时才有效——例如,一个容量有限且当前已满的 `QueueChannel`。 默认是无限。 <1> 此网关接收请求消息的通道。 <1> 对发送请求消息的 `Destination` 的引用。 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name` 之一是必需的。 你只能使用这三个属性中的一个。 <1> 一个 SpEL 表达式,评估为一个 `Destination`,请求消息将发送到该 `Destination`。 表达式可以生成 `Destination` 对象或 `String`。 它由 `DestinationResolver` 用于解析实际的 `Destination`。 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name` 之一是必需的。 你只能使用这三个属性中的一个。 <1> 发送请求消息的目的地名称。 它由 `DestinationResolver` 用于解析实际的 `Destination`。 `reply-destination`、`reply-destination-expression` 或 `reply-destination-name` 之一是必需的。 你只能使用这三个属性中的一个。 <1> 当设置为 `true` 时,表示由 `DestinationResolver` 解析的任何请求 `Destination` 都应该是 `Topic` 而不是 `Queue`。 <1> 指定消息的存活时间。 此设置仅在 `explicit-qos-enabled` 为 `true` 时生效。 <1> 指定此出站网关是否必须返回非 null 值。 默认情况下,此值为 `true`,当底层服务在 `receive-timeout` 后未返回值时,将抛出 `MessageTimeoutException`。 请注意,如果服务永远不期望返回回复,则最好使用 `<int-jms:outbound-channel-adapter/>` 而不是 `requires-reply="false"` 的 `<int-jms:outbound-gateway/>`。 后者会阻塞发送线程,等待 `receive-timeout` 期间的回复。 <1> 当使用 `<reply-listener />` 时,其生命周期(启动和停止)默认与网关匹配。 当此值大于 `0` 时,容器会按需启动(当发送请求时)。 容器会持续运行,直到至少此时间过去,且没有收到请求(并且没有未完成的回复)。 容器会在下一个请求时再次启动。 停止时间是最小值,实际上可能高达此值的 1.5 倍。 <1> 请参阅 <<Async Gateway,jms-async-gateway>>。 <1> 当包含此元素时,回复由异步 `MessageListenerContainer` 接收,而不是为每个回复创建一个消费者。 在许多情况下,这可能更高效。
将消息头映射到 JMS 消息和从 JMS 消息映射
JMS 消息可以包含元信息,例如 JMS API 头和简单属性。
你可以使用 JmsHeaderMapper
将它们映射到 Spring Integration 消息头以及从 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。
要启用此功能,你必须使用 DynamicJmsTemplate
配置出站端点,并将其 explicitQosEnabled
属性设置为 true。
Spring Integration Java DSL 默认配置 DynamicJmsTemplate
,但你仍然必须设置 explicitQosEnabled
属性。
从 4.0 版本开始,JMSPriority
头部映射到入站消息的标准 priority
头部。
以前,priority
头部仅用于出站消息。
要恢复到以前的行为(即不映射入站优先级),请将 DefaultJmsHeaderMapper
的 mapInboundPriority
属性设置为 false
。
从 4.3 版本开始,DefaultJmsHeaderMapper
将标准 correlationId
头部映射为消息属性,通过调用其 toString()
方法(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 属性分别映射到 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 消息。
请注意,在这种情况下,生产者和消费者事务是独立的。
这与事务上下文通过没有 <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
设置为 true
以进行共享订阅(需要 JMS 2.0 代理,并自版本 4.2 起可用)。
使用 subscription
命名订阅。
使用 JMS 消息选择器
使用 JMS 消息选择器,你可以根据 JMS 头和 JMS 属性筛选 JMS 消息。
例如,如果你想监听自定义 JMS 头属性 myHeaderProperty
等于 something
的消息,你可以指定以下表达式:
myHeaderProperty = 'something'
消息选择器表达式是 SQL-92 条件表达式语法的一个子集,并作为 Java Message Service 规范的一部分进行定义。
你可以使用 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://localhost:61616
(而不是 vm://localhost
)。
这两个示例都接受来自 stdin 的输入并回显到 stdout。
查看配置以了解这些消息如何通过 JMS 路由。