Annotation-driven Listener Endpoints

메시지를 비동기적으로 수신하는 가장 쉬운 방법은 annotation 기반 listener endpoint 인프라스트럭처를 사용하는 것입니다. 간단히 말해, 이는 managed bean의 메서드를 JMS listener endpoint로 노출할 수 있게 해 줍니다. 다음 예제는 이를 사용하는 방법을 보여 줍니다:

1@Component
2public class MyService {
3
4	@JmsListener(destination = "myDestination")
5	public void processOrder(String data) { ... }
6}

앞선 예제의 개념은 jakarta.jms.Destination myDestination에 메시지가 사용 가능해질 때마다 processOrder 메서드가 그에 따라 호출된다는 것입니다 (이 경우 JMS 메시지의 내용을 사용하며, 이는 MessageListenerAdapter 가 제공하는 것과 유사합니다).

annotated endpoint 인프라스트럭처는 각 annotated 메서드에 대해 JmsListenerContainerFactory를 사용하여 뒤에서 message listener 컨테이너를 생성합니다. 이러한 컨테이너는 애플리케이션 컨텍스트에 등록되지 않지만, JmsListenerEndpointRegistry bean을 사용하여 관리 목적을 위해 쉽게 찾을 수 있습니다.

@JmsListener는 Java 8에서 repeatable annotation이므로, 동일한 메서드에 여러 JMS destination을 연결하려면 여기에 추가적인 @JmsListener 선언을 추가하면 됩니다.

Enable Listener Endpoint Annotations

@JmsListener annotation 지원을 활성화하려면, 다음 예제와 같이 하나의 @Configuration 클래스에 @EnableJms를 추가하면 됩니다:

1@Configuration
2@EnableJms
3public class JmsConfiguration {
4
5	@Bean
6	public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory,
7			DestinationResolver destinationResolver) {
8
9		DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
10		factory.setConnectionFactory(connectionFactory);
11		factory.setDestinationResolver(destinationResolver);
12		factory.setSessionTransacted(true);
13		factory.setConcurrency("3-10");
14		return factory;
15	}
16}

1@Configuration
2@EnableJms
3class JmsConfiguration {
4
5	@Bean
6	fun jmsListenerContainerFactory(connectionFactory: ConnectionFactory, destinationResolver: DestinationResolver) =
7		DefaultJmsListenerContainerFactory().apply {
8			setConnectionFactory(connectionFactory)
9			setDestinationResolver(destinationResolver)
10			setSessionTransacted(true)
11			setConcurrency("3-10")
12		}
13}

1<jms:annotation-driven/>
2
3<bean id="jmsListenerContainerFactory" class="org.springframework.jms.config.DefaultJmsListenerContainerFactory">
4	<property name="connectionFactory" ref="connectionFactory"/>
5	<property name="destinationResolver" ref="destinationResolver"/>
6	<property name="sessionTransacted" value="true"/>
7	<property name="concurrency" value="3-10"/>
8</bean>

기본적으로, 인프라스트럭처는 message listener 컨테이너를 생성하기 위해 사용할 factory의 소스로 jmsListenerContainerFactory라는 이름의 bean을 찾습니다. 이 경우 (그리고 JMS 인프라스트럭처 설정은 무시하고), 세 개의 thread core pool size와 열 개의 thread maximum pool size로 processOrder 메서드를 호출할 수 있습니다.

각 annotation에 대해 사용할 listener 컨테이너 factory를 사용자 정의할 수 있으며, JmsListenerConfigurer 인터페이스를 구현하여 명시적인 기본값을 구성할 수도 있습니다. 기본값은 적어도 하나의 endpoint가 특정 컨테이너 factory 없이 등록된 경우에만 필요합니다. 자세한 내용과 예제는 JmsListenerConfigurer 를 구현하는 클래스의 javadoc을 참조하십시오.

Programmatic Endpoint Registration

JmsListenerEndpoint는 JMS endpoint의 모델을 제공하며, 해당 모델에 대한 컨테이너 구성을 담당합니다. 인프라스트럭처는 JmsListener annotation으로 감지되는 endpoint에 추가하여 endpoint를 프로그래밍 방식으로 구성할 수 있게 해 줍니다. 다음 예제는 그 방법을 보여 줍니다:

1@Configuration
2@EnableJms
3public class AppConfig implements JmsListenerConfigurer {
4
5	@Override
6	public void configureJmsListeners(JmsListenerEndpointRegistrar registrar) {
7		SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
8		endpoint.setId("myJmsEndpoint");
9		endpoint.setDestination("anotherQueue");
10		endpoint.setMessageListener(message -> {
11			// processing
12		});
13		registrar.registerEndpoint(endpoint);
14	}
15}

앞선 예제에서는 실제로 호출할 MessageListener를 제공하는 SimpleJmsListenerEndpoint를 사용했습니다. 그러나 custom invocation mechanism을 설명하기 위해 자체 endpoint variant를 만들 수도 있습니다.

@JmsListener 사용을 완전히 생략하고 JmsListenerConfigurer를 통해 endpoint만 프로그래밍 방식으로 등록할 수도 있습니다.

Annotated Endpoint Method Signature

지금까지는 endpoint에 단순한 String을 주입했지만, 실제로는 매우 유연한 메서드 시그니처를 가질 수 있습니다. 다음 예제에서는 custom header와 함께 Order를 주입하도록 이를 다시 작성합니다:

1@Component
2public class MyService {
3
4	@JmsListener(destination = "myDestination")
5	public void processOrder(Order order, @Header("order_type") String orderType) {
6		...
7	}
8}

JMS listener endpoint에 주입할 수 있는 주요 요소는 다음과 같습니다:

• raw jakarta.jms.Message 또는 그 하위 클래스 (단, 수신 메시지 타입과 일치해야 함).
• native JMS API에 대한 선택적 접근(예: custom reply 전송)을 위한 jakarta.jms.Session.
• 수신 JMS 메시지를 나타내는 org.springframework.messaging.Message. 이 메시지는 JmsHeaders에 의해 정의된 custom header와 standard header를 모두 보유합니다.
• standard JMS header를 포함하여 특정 header 값을 추출하기 위한 @Header가 부착된 메서드 argument.
• 모든 header에 접근하기 위해 java.util.Map에 할당 가능해야 하는 @Headers가 부착된 argument.
• 지원되는 타입 (Message 또는 Session)이 아닌 annotation이 없는 element는 payload로 간주됩니다. parameter에 @Payload annotation을 부착하여 이를 명시적으로 만들 수 있습니다. 또한 @Valid를 추가하여 validation을 활성화할 수도 있습니다.

Spring의 Message 추상화를 주입하는 기능은 전송 방식에 특화된 API에 의존하지 않고 전송 방식에 특화된 메시지에 저장된 모든 정보의 이점을 활용하는 데 특히 유용합니다. 다음 예제는 이를 수행하는 방법을 보여 줍니다:

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

메서드 argument 처리는 DefaultMessageHandlerMethodFactory에 의해 제공되며, 이를 추가 메서드 argument를 지원하도록 추가로 사용자 정의할 수 있습니다. 변환 및 검증 지원도 그곳에서 사용자 정의할 수 있습니다.

예를 들어, Order가 처리되기 전에 유효한지 확인하려면, payload에 @Valid annotation을 부착하고 필요한 validator를 구성하면 됩니다. 다음 예제는 이를 보여 줍니다:

1@Configuration
2@EnableJms
3public class AppConfig implements JmsListenerConfigurer {
4
5	@Override
6	public void configureJmsListeners(JmsListenerEndpointRegistrar registrar) {
7		registrar.setMessageHandlerMethodFactory(myJmsHandlerMethodFactory());
8	}
9
10	@Bean
11	public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
12		DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
13		factory.setValidator(myValidator());
14		return factory;
15	}
16}

Response Management

MessageListenerAdapter 에 있는 기존 지원은 이미 메서드가 non-void 반환 타입을 갖도록 허용합니다. 그런 경우 호출 결과는 jakarta.jms.Message에 캡슐화되며, 원본 메시지의 JMSReplyTo header에 지정된 destination 또는 listener에 구성된 기본 destination 중 하나로 전송됩니다. 이제 messaging 추상화의 @SendTo annotation을 사용하여 해당 기본 destination을 설정할 수 있습니다.

processOrder 메서드가 이제 OrderStatus를 반환해야 한다고 가정하면, 다음 예제와 같이 자동으로 response를 전송하도록 작성할 수 있습니다:

1@JmsListener(destination = "myDestination")
2@SendTo("status")
3public OrderStatus processOrder(Order order) {
4	// order processing
5	return status;
6}

여러 개의 @JmsListener annotation이 부착된 메서드가 있는 경우, 클래스 수준에 @SendTo annotation을 배치하여 기본 reply destination을 공유할 수도 있습니다.

전송 방식에 독립적인 방식으로 추가 header를 설정해야 하는 경우, 다음과 같은 메서드로 Message를 대신 반환할 수 있습니다:

1@JmsListener(destination = "myDestination")
2@SendTo("status")
3public Message<OrderStatus> processOrder(Order order) {
4	// order processing
5	return MessageBuilder
6			.withPayload(status)
7			.setHeader("code", 1234)
8			.build();
9}

runtime에 response destination을 계산해야 하는 경우, runtime에 사용할 destination도 제공하는 JmsResponse 인스턴스에 response를 캡슐화할 수 있습니다. 이전 예제를 다음과 같이 다시 작성할 수 있습니다:

1@JmsListener(destination = "myDestination")
2public JmsResponse<Message<OrderStatus>> processOrder(Order order) {
3	// order processing
4	Message<OrderStatus> response = MessageBuilder
5			.withPayload(status)
6			.setHeader("code", 1234)
7			.build();
8	return JmsResponse.forQueue(response, "status");
9}

마지막으로, priority나 time to live과 같은 response에 대한 QoS 값을 지정해야 하는 경우, 다음 예제와 같이 JmsListenerContainerFactory를 적절히 구성할 수 있습니다:

1@Configuration
2@EnableJms
3public class AppConfig {
4
5	@Bean
6	public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
7		DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
8		factory.setConnectionFactory(connectionFactory());
9		QosSettings replyQosSettings = new QosSettings();
10		replyQosSettings.setPriority(2);
11		replyQosSettings.setTimeToLive(10000);
12		factory.setReplyQosSettings(replyQosSettings);
13		return factory;
14	}
15}