Receiving a Message

이 문서는 Spring에서 JMS로 메시지를 수신하는 방법을 설명합니다.

Synchronous Receipt

JMS는 일반적으로 비동기 처리와 연관되어 있지만, 메시지를 동기적으로 소비할 수도 있습니다. JmsTemplate 및 JmsClient의 receive(..) 메서드는 이 기능을 제공합니다.

동기 수신 동안에는 호출한 스레드가 메시지가 사용 가능해질 때까지 블록됩니다. 이는 호출한 스레드가 잠재적으로 무기한 블록될 수 있기 때문에 위험한 작업이 될 수 있습니다. receiveTimeout 프로퍼티는 수신기가 메시지를 기다리는 것을 포기하기 전에 얼마나 오래 기다려야 하는지를 지정합니다.

Asynchronous Receipt: Message-Driven POJOs

Spring은 @JmsListener<br>어노테이션을 사용하여 어노테이션 기반 리스너 엔드포인트도 지원하며 엔드포인트를 프로그래밍 방식으로 등록할 수 있는 개방형 인프라스트럭처를 제공합니다.<br>이는 비동기 수신기를 설정하는 가장 편리한 방법입니다.<br>자세한 내용은 Enable Listener Endpoint Annotations를 참조하십시오.

EJB 세계의 Message-Driven Bean(MDB)과 유사한 방식으로, Message-Driven POJO(MDP)는 JMS 메시지의 수신기 역할을 합니다. MDP에 대한 한 가지 제약 사항은 (단, Using MessageListenerAdapter를 참조하십시오) jakarta.jms.MessageListener 인터페이스를 구현해야 한다는 점입니다.

POJO가 여러 스레드에서 메시지를 수신하는 경우에는 구현이 스레드 세이프하도록 보장하는 것이 중요하다는 점에 유의하십시오.

다음 예제는 MDP의 간단한 구현을 보여 줍니다:

1public class ExampleListener implements MessageListener {
2
3	public void onMessage(Message message) {
4		if (message instanceof TextMessage textMessage) {
5			try {
6				System.out.println(textMessage.getText());
7			}
8			catch (JMSException ex) {
9				throw new RuntimeException(ex);
10			}
11		}
12		else {
13			throw new IllegalArgumentException("Message must be of type TextMessage");
14		}
15	}
16}

1class ExampleListener : MessageListener {
2
3	override fun onMessage(message: Message) {
4		if (message is TextMessage) {
5			try {
6				println(message.text)
7			} catch (ex: JMSException) {
8				throw RuntimeException(ex)
9			}
10		} else {
11			throw IllegalArgumentException("Message must be of type TextMessage")
12		}
13	}
14}

MessageListener를 구현했으면 이제 메시지 리스너 컨테이너를 생성할 차례입니다.

다음 예제는 Spring에 포함된 메시지 리스너 컨테이너 중 하나 (이 경우 DefaultMessageListenerContainer)를 정의하고 구성하는 방법을 보여 줍니다:

1@Bean
2ExampleListener messageListener() {
3	return new ExampleListener();
4}
5
6@Bean
7DefaultMessageListenerContainer jmsContainer(ConnectionFactory connectionFactory, Destination destination,
8		ExampleListener messageListener) {
9
10	DefaultMessageListenerContainer jmsContainer = new DefaultMessageListenerContainer();
11	jmsContainer.setConnectionFactory(connectionFactory);
12	jmsContainer.setDestination(destination);
13	jmsContainer.setMessageListener(messageListener);
14	return jmsContainer;
15}

1@Bean
2fun messageListener() = ExampleListener()
3
4@Bean
5fun jmsContainer(connectionFactory: ConnectionFactory, destination: Destination, messageListener: ExampleListener) =
6	DefaultMessageListenerContainer().apply {
7		setConnectionFactory(connectionFactory)
8		setDestination(destination)
9		setMessageListener(messageListener)
10	}

1<!-- this is the Message Driven POJO (MDP) -->
2<bean id="messageListener" class="jmsexample.ExampleListener"/>
3
4<!-- and this is the message listener container -->
5<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
6	<property name="connectionFactory" ref="connectionFactory"/>
7	<property name="destination" ref="destination"/>
8	<property name="messageListener" ref="messageListener"/>
9</bean>

각 구현에서 지원하는 기능에 대한 전체 설명은 Spring javadoc의 다양한 메시지 리스너 컨테이너 (모두 MessageListenerContainer를 구현함)를 참조하십시오.

Using the SessionAwareMessageListener Interface

SessionAwareMessageListener 인터페이스는 JMS MessageListener 인터페이스와 유사한 계약을 제공하지만 메시지 처리 메서드가 Message가 수신된 JMS Session에 접근할 수 있도록 해 주는 Spring 전용 인터페이스입니다. 다음 목록은 SessionAwareMessageListener 인터페이스의 정의를 보여 줍니다:

1package org.springframework.jms.listener;
2
3public interface SessionAwareMessageListener {
4
5	void onMessage(Message message, Session session) throws JMSException;
6}

MDP가 수신된 메시지에 응답할 수 있기를 원한다면 (즉, onMessage(Message, Session) 메서드에 제공된 Session을 사용하여) 표준 JMS MessageListener 인터페이스 대신 이 인터페이스를 MDP에 구현하도록 선택할 수 있습니다. Spring에 포함된 모든 메시지 리스너 컨테이너 구현은 MessageListener 또는 SessionAwareMessageListener 인터페이스를 구현하는 MDP를 지원합니다.

SessionAwareMessageListener를 구현하는 클래스는 그 인터페이스를 통해 Spring에 묶이게 된다는 단서가 붙습니다. 이를 사용할지 여부에 대한 선택은 애플리케이션 개발자 또는 아키텍트인 여러분에게 전적으로 맡겨져 있습니다.

SessionAwareMessageListener 인터페이스의 onMessage(..) 메서드는 JMSException을 던진다는 점에 유의하십시오. 표준 JMS MessageListener 인터페이스와는 달리, SessionAwareMessageListener 인터페이스를 사용할 때는 던져진 예외를 처리하는 책임이 클라이언트 코드에 있습니다.

Using MessageListenerAdapter

MessageListenerAdapter 클래스는 Spring의 비동기 메시징 지원에서 마지막 컴포넌트입니다. 요약하면, 이 클래스는 거의 모든 클래스를 MDP로 노출할 수 있게 해 줍니다(일부 제약은 있습니다).

다음 인터페이스 정의를 살펴보십시오:

1public interface MessageDelegate {
2
3	void handleMessage(String message);
4
5	void handleMessage(Map message);
6
7	void handleMessage(byte[] message);
8
9	void handleMessage(Serializable message);
10}

1interface MessageDelegate {
2	fun handleMessage(message: String)
3	fun handleMessage(message: Map<*, *>)
4	fun handleMessage(message: ByteArray)
5	fun handleMessage(message: Serializable)
6}

인터페이스가 MessageListener나 SessionAwareMessageListener 인터페이스를 전혀 확장하지 않음에도 불구하고, MessageListenerAdapter 클래스를 사용하여 여전히 이를 MDP로 사용할 수 있다는 점에 주목하십시오. 또한 다양한 메시지 처리 메서드가 수신하고 처리할 수 있는 다양한 Message 타입의 내용에 따라 강하게 타입이 지정되어 있다는 점에도 주목하십시오.

이제 MessageDelegate 인터페이스의 다음 구현을 살펴보십시오:

1public class DefaultMessageDelegate implements MessageDelegate {
2
3	@Override
4	public void handleMessage(String message) {
5		// ...
6	}
7
8	@Override
9	public void handleMessage(Map message) {
10		// ...
11	}
12
13	@Override
14	public void handleMessage(byte[] message) {
15		// ...
16	}
17
18	@Override
19	public void handleMessage(Serializable message) {
20		// ...
21	}
22}

1class DefaultMessageDelegate : MessageDelegate {
2
3	override fun handleMessage(message: String) {
4		// ...
5	}
6
7	override fun handleMessage(message: Map<*, *>) {
8		// ...
9	}
10
11	override fun handleMessage(message: ByteArray) {
12		// ...
13	}
14
15	override fun handleMessage(message: Serializable) {
16		// ...
17	}
18}

특히, 앞의 MessageDelegate 인터페이스 구현 (DefaultMessageDelegate 클래스)이 JMS에 전혀 의존하지 않는다는 점에 주목하십시오. 이는 우리가 다음과 같은 설정을 통해 MDP로 만들 수 있는 진정한 POJO입니다:

1@Bean
2MessageListenerAdapter messageListener(DefaultMessageDelegate messageDelegate) {
3	return new MessageListenerAdapter(messageDelegate);
4}
5
6@Bean
7DefaultMessageListenerContainer jmsContainer(ConnectionFactory connectionFactory, Destination destination,
8		ExampleListener messageListener) {
9
10	DefaultMessageListenerContainer jmsContainer = new DefaultMessageListenerContainer();
11	jmsContainer.setConnectionFactory(connectionFactory);
12	jmsContainer.setDestination(destination);
13	jmsContainer.setMessageListener(messageListener);
14	return jmsContainer;
15}

1@Bean
2fun messageListener(messageDelegate: DefaultMessageDelegate): MessageListenerAdapter {
3	return MessageListenerAdapter(messageDelegate)
4}
5
6@Bean
7fun jmsContainer(connectionFactory: ConnectionFactory, destination: Destination, messageListener: ExampleListener) =
8	DefaultMessageListenerContainer().apply {
9		setConnectionFactory(connectionFactory)
10		setDestination(destination)
11		setMessageListener(messageListener)
12	}

1<!-- this is the Message Driven POJO (MDP) -->
2<bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
3	<constructor-arg>
4		<bean class="jmsexample.DefaultMessageDelegate"/>
5	</constructor-arg>
6</bean>
7
8<!-- and this is the message listener container... -->
9<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
10	<property name="connectionFactory" ref="connectionFactory"/>
11	<property name="destination" ref="destination"/>
12	<property name="messageListener" ref="messageListener"/>
13</bean>

다음 예제는 JMS TextMessage 메시지만 수신할 수 있는 또 다른 MDP를 보여 줍니다. 메시지 처리 메서드의 이름이 실제로는 receive라는 점에 주의하십시오(MessageListenerAdapter에서 메시지 처리 메서드의 이름은 기본적으로 handleMessage이지만, 이 섹션의 뒷부분에서 볼 수 있듯이 구성 가능합니다). 또한 receive(..) 메서드가 JMS TextMessage 메시지만 수신하고 응답하도록 강하게 타입이 지정되어 있다는 점에도 주목하십시오.

다음 목록은 TextMessageDelegate 인터페이스의 정의를 보여 줍니다:

1public interface TextMessageDelegate {
2
3	void receive(TextMessage message);
4}

1interface TextMessageDelegate {
2	fun receive(message: TextMessage)
3}

다음 목록은 TextMessageDelegate 인터페이스를 구현하는 클래스를 보여 줍니다:

1public class DefaultTextMessageDelegate implements TextMessageDelegate {
2
3	@Override
4	public void receive(TextMessage message) {
5		// ...
6	}
7}

1class DefaultTextMessageDelegate : TextMessageDelegate {
2
3	override fun receive(message: TextMessage) {
4		// ...
5	}
6}

해당 MessageListenerAdapter의 설정은 다음과 같습니다:

1@Bean
2MessageListenerAdapter messageListener(DefaultTextMessageDelegate messageDelegate) {
3	MessageListenerAdapter messageListener = new MessageListenerAdapter(messageDelegate);
4	messageListener.setDefaultListenerMethod("receive");
5	// We don't want automatic message context extraction
6	messageListener.setMessageConverter(null);
7	return messageListener;
8}

1@Bean
2fun messageListener(messageDelegate: DefaultTextMessageDelegate) = MessageListenerAdapter(messageDelegate).apply {
3	setDefaultListenerMethod("receive")
4	// We don't want automatic message context extraction
5	setMessageConverter(null)
6}

1<bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
2	<constructor-arg>
3		<bean class="jmsexample.DefaultTextMessageDelegate"/>
4	</constructor-arg>
5	<property name="defaultListenerMethod" value="receive"/>
6	<!-- we don't want automatic message context extraction -->
7	<property name="messageConverter">
8		<null/>
9	</property>
10</bean>

messageListener가 TextMessage가 아닌 타입의 JMS Message를 수신하면 IllegalStateException이 던져지고(그 후에 삼켜진다는) 점에 유의하십시오. MessageListenerAdapter 클래스의 또 다른 기능은 핸들러 메서드가 non-void 값을 반환할 경우 자동으로 응답 Message를 되돌려 보낼 수 있는 능력입니다.

다음 인터페이스와 클래스를 살펴보십시오:

1public interface ResponsiveTextMessageDelegate {
2
3	// Notice the return type...
4	String receive(TextMessage message);
5}

1interface ResponsiveTextMessageDelegate {
2
3	// Notice the return type...
4	fun receive(message: TextMessage): String
5}

1public class DefaultResponsiveTextMessageDelegate implements ResponsiveTextMessageDelegate {
2
3	@Override
4	public String receive(TextMessage message) {
5		return "message";
6	}
7}

1class DefaultResponsiveTextMessageDelegate : ResponsiveTextMessageDelegate {
2
3	override fun receive(message: TextMessage): String {
4		return "message"
5	}
6}

DefaultResponsiveTextMessageDelegate를 MessageListenerAdapter와 함께 사용하면, 'receive(..)' 메서드 실행에서 반환되는 non-null 값은 (기본 설정에서) TextMessage로 변환됩니다. 결과 TextMessage는 원래 Message의 JMS Reply-To 프로퍼티에 정의된 Destination(존재하는 경우) 또는 MessageListenerAdapter에 설정된 기본 Destination(구성된 경우)으로 전송됩니다.

Destination을 찾을 수 없으면 InvalidDestinationException이 던져집니다(이 예외는 삼켜지지 않고 호출 스택을 따라 전파된다는 점에 유의하십시오).

Processing Messages Within Transactions

트랜잭션 내에서 메시지 리스너를 호출하려면 리스너 컨테이너를 재구성하기만 하면 됩니다.

리스너 컨테이너 정의의 sessionTransacted 플래그를 통해 로컬 리소스 트랜잭션을 활성화할 수 있습니다. 각 메시지 리스너 호출은 활성화된 JMS 트랜잭션 내에서 동작하며, 리스너 실행 실패 시 메시지 수신이 롤백됩니다.

응답 메시지를 보내는 작업(SessionAwareMessageListener를 통해)은 같은 로컬 트랜잭션의 일부이지만, 그 외의 리소스 작업(예: 데이터베이스 액세스)은 독립적으로 동작합니다. 이는 일반적으로 리스너 구현에서 중복 메시지 감지가 필요함을 의미하는데, 이는 데이터베이스 처리는 커밋되었지만 메시지 처리가 커밋에 실패한 경우를 처리하기 위함입니다.

다음 빈 정의를 살펴보십시오:

1@Bean
2DefaultMessageListenerContainer jmsContainer(ConnectionFactory connectionFactory, Destination destination,
3		ExampleListener messageListener) {
4
5	DefaultMessageListenerContainer jmsContainer = new DefaultMessageListenerContainer();
6	jmsContainer.setConnectionFactory(connectionFactory);
7	jmsContainer.setDestination(destination);
8	jmsContainer.setMessageListener(messageListener);
9	jmsContainer.setSessionTransacted(true);
10	return jmsContainer;
11}

1@Bean
2fun jmsContainer(connectionFactory: ConnectionFactory, destination: Destination, messageListener: ExampleListener) =
3	DefaultMessageListenerContainer().apply {
4		setConnectionFactory(connectionFactory)
5		setDestination(destination)
6		setMessageListener(messageListener)
7		isSessionTransacted = true
8	}

1<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
2	<property name="connectionFactory" ref="connectionFactory"/>
3	<property name="destination" ref="destination"/>
4	<property name="messageListener" ref="messageListener"/>
5	<property name="sessionTransacted" value="true"/>
6</bean>

외부에서 관리되는 트랜잭션에 참여하려면 트랜잭션 매니저를 구성하고 외부에서 관리되는 트랜잭션을 지원하는 리스너 컨테이너(일반적으로 DefaultMessageListenerContainer)를 사용해야 합니다.

메시지 리스너 컨테이너를 XA 트랜잭션 참여용으로 구성하려면 기본적으로 Jakarta EE 서버의 트랜잭션 서브시스템에 위임하는 JtaTransactionManager를 구성해야 합니다. 기본 JMS ConnectionFactory가 XA-capable이어야 하고 JTA 트랜잭션 코디네이터에 적절히 등록되어야 한다는 점에 유의하십시오(Jakarta EE 서버의 JNDI 리소스 설정을 확인하십시오).

이를 통해 메시지 수신과 (예를 들어) 데이터베이스 액세스가 동일한 트랜잭션의 일부가 되도록 할 수 있습니다(unified commit semantics를 제공하는 대신 XA 트랜잭션 로그 오버헤드가 발생합니다).

다음 빈 정의는 트랜잭션 매니저를 생성합니다:

1@Bean
2JtaTransactionManager transactionManager()  {
3	return new JtaTransactionManager();
4}

1@Bean
2fun transactionManager() = JtaTransactionManager()

1<bean id="transactionManager" class="org.springframework.transaction.jta.JtaTransactionManager"/>

그런 다음 앞에서 사용한 컨테이너 설정에 이를 추가해야 합니다. 컨테이너가 나머지를 처리합니다. 다음 예제는 이를 수행하는 방법을 보여 줍니다:

1@Bean
2DefaultMessageListenerContainer jmsContainer(ConnectionFactory connectionFactory, Destination destination,
3		ExampleListener messageListener) {
4
5	DefaultMessageListenerContainer jmsContainer = new DefaultMessageListenerContainer();
6	jmsContainer.setConnectionFactory(connectionFactory);
7	jmsContainer.setDestination(destination);
8	jmsContainer.setMessageListener(messageListener);
9	jmsContainer.setSessionTransacted(true);
10	return jmsContainer;
11}

1@Bean
2fun jmsContainer(connectionFactory: ConnectionFactory, destination: Destination, messageListener: ExampleListener,
3				 transactionManager: JtaTransactionManager) =
4	DefaultMessageListenerContainer().apply {
5		setConnectionFactory(connectionFactory)
6		setDestination(destination)
7		setMessageListener(messageListener)
8		setTransactionManager(transactionManager)
9	}

1<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
2	<property name="connectionFactory" ref="connectionFactory"/>
3	<property name="destination" ref="destination"/>
4	<property name="messageListener" ref="messageListener"/>
5	<property name="transactionManager" ref="transactionManager"/>
6</bean>