注解驱动的监听器端点

异步接收消息最简单的方法是使用注解驱动的监听器端点基础设施。简而言之，它允许你将一个托管 bean 的方法暴露为 JMS 监听器端点。以下示例展示了如何使用它：

@Component
public class MyService {

	@JmsListener(destination = "myDestination")
	public void processOrder(String data) { ... }
}

前述示例的目的是，每当 jakarta.jms.Destination myDestination 上有消息可用时，processOrder 方法就会相应地被调用（在这种情况下，使用 JMS 消息的内容，类似于 MessageListenerAdapter 所提供的功能）。

注解端点基础设施在幕后为每个注解方法创建了一个消息监听器容器，它通过 JmsListenerContainerFactory 实现。这样的容器不会注册到应用程序上下文，但可以通过 JmsListenerEndpointRegistry bean 轻松定位，以便进行管理。

@JmsListener 在 Java 8 中是可重复的注解，因此你可以通过添加额外的 @JmsListener 声明将多个 JMS 目标与同一个方法关联起来。

启用监听器端点注解

要启用 @JmsListener 注解支持，你可以在其中一个 @Configuration 类中添加 @EnableJms，如以下示例所示：

  • Java

  • Kotlin

  • Xml

@Configuration
@EnableJms
public class JmsConfiguration {

	@Bean
	public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory,
			DestinationResolver destinationResolver) {

		DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
		factory.setConnectionFactory(connectionFactory);
		factory.setDestinationResolver(destinationResolver);
		factory.setSessionTransacted(true);
		factory.setConcurrency("3-10");
		return factory;
	}
}
@Configuration
@EnableJms
class JmsConfiguration {

	@Bean
	fun jmsListenerContainerFactory(connectionFactory: ConnectionFactory, destinationResolver: DestinationResolver) =
		DefaultJmsListenerContainerFactory().apply {
			setConnectionFactory(connectionFactory)
			setDestinationResolver(destinationResolver)
			setSessionTransacted(true)
			setConcurrency("3-10")
		}
}
<jms:annotation-driven/>

<bean id="jmsListenerContainerFactory" class="org.springframework.jms.config.DefaultJmsListenerContainerFactory">
	<property name="connectionFactory" ref="connectionFactory"/>
	<property name="destinationResolver" ref="destinationResolver"/>
	<property name="sessionTransacted" value="true"/>
	<property name="concurrency" value="3-10"/>
</bean>

默认情况下，基础设施会查找名为 jmsListenerContainerFactory 的 bean 作为创建消息监听器容器的工厂来源。在这种情况下（并且忽略 JMS 基础设施设置），你可以使用三个线程的核心池大小和十个线程的最大池大小来调用 processOrder 方法。

你可以自定义每个注解使用的监听器容器工厂，或者通过实现 JmsListenerConfigurer 接口来配置一个显式默认值。仅当至少一个端点未指定容器工厂就注册时，才需要默认值。有关详细信息和示例，请参阅实现 JmsListenerConfigurer 的类的 javadoc。

编程方式注册端点

JmsListenerEndpoint 提供了 JMS 端点的模型，并负责为该模型配置容器。除了通过 JmsListener 注解检测到的端点之外，该基础设施还允许你以编程方式配置端点。以下示例展示了如何实现：

@Configuration
@EnableJms
public class AppConfig implements JmsListenerConfigurer {

	@Override
	public void configureJmsListeners(JmsListenerEndpointRegistrar registrar) {
		SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
		endpoint.setId("myJmsEndpoint");
		endpoint.setDestination("anotherQueue");
		endpoint.setMessageListener(message -> {
			// processing
		});
		registrar.registerEndpoint(endpoint);
	}
}

在前面的示例中，我们使用了 SimpleJmsListenerEndpoint，它提供了要调用的实际 MessageListener。然而，你也可以构建自己的端点变体来描述自定义调用机制。

请注意，你可以完全跳过使用 @JmsListener，只通过 JmsListenerConfigurer 以编程方式注册你的端点。

注解端点方法签名

到目前为止，我们一直在端点中注入一个简单的 String，但它实际上可以有一个非常灵活的方法签名。在以下示例中，我们将其重写为注入带有自定义头信息的 Order：

@Component
public class MyService {

	@JmsListener(destination = "myDestination")
	public void processOrder(Order order, @Header("order_type") String orderType) {
		...
	}
}

你可以在 JMS 监听器端点中注入的主要元素如下：

  • 原始的 jakarta.jms.Message 或其任何子类（前提是它与传入消息类型匹配）。

  • jakarta.jms.Session，用于可选地访问原生 JMS API（例如，用于发送自定义回复）。

  • 表示传入 JMS 消息的 org.springframework.messaging.Message。请注意，此消息包含自定义和标准头信息（由 JmsHeaders 定义）。

  • 使用 @Header 注解的方法参数，用于提取特定的头值，包括标准 JMS 头。

  • 一个使用 @Headers 注解的参数，它还必须可赋值给 java.util.Map，以便访问所有头。

  • 未注解且不是受支持类型（Message 或 Session）的元素被视为有效负载。你可以通过使用 @Payload 注解参数来明确这一点。你还可以通过添加额外的 @Valid 来启用验证。

注入 Spring 的 Message 抽象的能力特别有用，可以利用存储在传输特定消息中的所有信息，而无需依赖传输特定 API。以下示例展示了如何实现：

@JmsListener(destination = "myDestination")
public void processOrder(Message<Order> order) { ... }

方法参数的处理由 DefaultMessageHandlerMethodFactory 提供，你可以进一步自定义它以支持额外的方​​法参数。你也可以在那里自定义转换和验证支持。

例如，如果我们要确保 Order 在处理之前有效，我们可以使用 @Valid 注解有效负载并配置必要的验证器，如以下示例所示：

@Configuration
@EnableJms
public class AppConfig implements JmsListenerConfigurer {

	@Override
	public void configureJmsListeners(JmsListenerEndpointRegistrar registrar) {
		registrar.setMessageHandlerMethodFactory(myJmsHandlerMethodFactory());
	}

	@Bean
	public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
		DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
		factory.setValidator(myValidator());
		return factory;
	}
}

响应管理

MessageListenerAdapter 中现有的支持已允许你的方法具有非 void 返回类型。在这种情况下，调用的结果被封装在一个 jakarta.jms.Message 中，发送到原始消息的 JMSReplyTo 头中指定的目标，或者发送到监听器上配置的默认目标。你现在可以使用消息抽象的 @SendTo 注解设置该默认目标。

假设我们的 processOrder 方法现在应该返回一个 OrderStatus，我们可以将其编写为自动发送响应，如以下示例所示：

@JmsListener(destination = "myDestination")
@SendTo("status")
public OrderStatus processOrder(Order order) {
	// order processing
	return status;
}
如果你有几个带有 @JmsListener 注解的方法，你也可以在类级别放置 @SendTo 注解，以共享一个默认的回复目标。

如果你需要以与传输无关的方式设置额外的头信息，你可以返回一个 Message，方法类似于以下内容：

@JmsListener(destination = "myDestination")
@SendTo("status")
public Message<OrderStatus> processOrder(Order order) {
	// order processing
	return MessageBuilder
			.withPayload(status)
			.setHeader("code", 1234)
			.build();
}

如果你需要在运行时计算响应目标，你可以将响应封装在一个 JmsResponse 实例中，该实例还提供运行时要使用的目标。我们可以将前面的示例重写如下：

@JmsListener(destination = "myDestination")
public JmsResponse<Message<OrderStatus>> processOrder(Order order) {
	// order processing
	Message<OrderStatus> response = MessageBuilder
			.withPayload(status)
			.setHeader("code", 1234)
			.build();
	return JmsResponse.forQueue(response, "status");
}

最后，如果你需要为响应指定一些 QoS 值，例如优先级或存活时间，你可以相应地配置 JmsListenerContainerFactory，如以下示例所示：

@Configuration
@EnableJms
public class AppConfig {

	@Bean
	public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
		DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
		factory.setConnectionFactory(connectionFactory());
		QosSettings replyQosSettings = new QosSettings();
		replyQosSettings.setPriority(2);
		replyQosSettings.setTimeToLive(10000);
		factory.setReplyQosSettings(replyQosSettings);
		return factory;
	}
}
© . This site is unofficial and not affiliated with VMware.