Synchronous Request Response with ActiveMQ and Spring | CodeDependents
Request Response with JMS
ActiveMQ documentation actually has a pretty good overview of how request-response semantics work in JMS.
You might think at first that to implement request-response type operations in JMS that you should create a new consumer with a selector per request; or maybe create a new temporary queue per request.
Creating temporary destinations, consumers, producers and connections are all synchronous request-response operations with the broker and so should be avoided for processing each request as it results in lots of chat with the JMS broker.
The best way to implement request-response over JMS is to create a temporary queue and consumer per client on startup, set JMSReplyTo property on each message to the temporary queue and then use a correlationID on each message to correlate request messages to response messages. This avoids the overhead of creating and closing a consumer for each request (which is expensive). It also means you can share the same producer & consumer across many threads if you want (or pool them maybe).
This is a pretty good start but it requires some tweaking to work best in Spring. It also should be noted that Lingoand Camel are also suggested as options when using ActiveMQ. In my previous post I addressed why I don’t use either of these options. In short Camel is more power than is needed for basic messaging and Lingo is built onJencks, neither of which have been updated in years.
Request Response in Spring
The first thing to notice is that its infeasible to create a consumer and temporary queue per client in Spring since pooling resources is required overcome the JmsTemplate gotchas. To get around this, I suggest using predefined request and response queues, removing the overhead of creating a temporary queue for each request/response. To allow for multiple consumers and producers on the same queue the JMSCorrelationId is used to correlated the request with its response message.
At this point I implemented the following naive solution:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
@Componentpublic class Requestor { private static final class CorrelationIdPostProcessor implements MessagePostProcessor { private final String correlationId; public CorrelationIdPostProcessor( final String correlationId ) { this.correlationId = correlationId; } @Override public Message postProcessMessage( final Message msg ) throws JMSException { msg.setJMSCorrelationID( correlationId ); return msg; } } private final JmsTemplate jmsTemplate; @Autowired public RequestGateway( JmsTemplate jmsTemplate ) { this.jmsTemplate = jmsTemplate; } public String request( final String request, String queue ) throws IOException { final String correlationId = UUID.randomUUID().toString(); jmsTemplate.convertAndSend( queue+".request", request, new CorrelationIdPostProcessor( correlationId ) ); return (String) jmsTemplate.receiveSelectedAndConvert( queue+".response", "JMSCorrelationID='" + correlationId + "'" ); }} |
This worked for a while until the system started occasionally timing out when making a request against a particularly fast responding service. After some debugging it became apparent that the service was responding so quickly that the receive() call was not fully initialized, causing it to miss the message. Once it finished initializing, it would wait until the timeout and fail. Unfortunately, there is very little in the way of documentation for this and the best suggestion I could find still seemed to leave open the possibility for the race condition by creating the consumer after sending the message. Luckily, according to the JMS spec, a consumer becomes active as soon as it is created and, assuming the connection has been started, it will start consuming messages. This allows for the reordering of the method calls leading to the slightly more verbose but also more correct solution. (NOTE: Thanks to Aaron Korver for pointing out that ProducerConsumer needs to implement SessionCallback and that true needs to be passed to the JmsTemplate.execute() for the connection to be started.)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
@Componentpublic class Requestor { private static final class ProducerConsumer implements SessionCallback<Message> { private static final int TIMEOUT = 5000; private final String msg; private final DestinationResolver destinationResolver; private final String queue; public ProducerConsumer( final String msg, String queue, final DestinationResolver destinationResolver ) { this.msg = msg; this.queue = queue; this.destinationResolver = destinationResolver; } public Message doInJms( final Session session ) throws JMSException { MessageConsumer consumer = null; MessageProducer producer = null; try { final String correlationId = UUID.randomUUID().toString(); final Destination requestQueue = destinationResolver.resolveDestinationName( session, queue+".request", false ); final Destination replyQueue = destinationResolver.resolveDestinationName( session, queue+".response", false ); // Create the consumer first! consumer = session.createConsumer( replyQueue, "JMSCorrelationID = '" + correlationId + "'" ); final TextMessage textMessage = session.createTextMessage( msg ); textMessage.setJMSCorrelationID( correlationId ); textMessage.setJMSReplyTo( replyQueue ); // Send the request second! producer = session.createProducer( requestQueue ); producer.send( requestQueue, textMessage ); // Block on receiving the response with a timeout return consumer.receive( TIMEOUT ); } finally { // Don't forget to close your resources JmsUtils.closeMessageConsumer( consumer ); JmsUtils.closeMessageProducer( producer ); } } } private final JmsTemplate jmsTemplate; @Autowired public Requestor( final JmsTemplate jmsTemplate ) { this.jmsTemplate = jmsTemplate; } public String request( final String request, String queue ) { // Must pass true as the second param to start the connection return (String) jmsTemplate.execute( new ProducerConsumer( msg, queue, jmsTemplate.getDestinationResolver() ), true ); }} |
About Pooling
Once the request/response logic was correct it was time to load test. Almost instantly, memory usage exploded and the garbage collector started thrashing. Inspecting ActiveMQ with the Web Console showed that MessageConsumers were hanging around even though they were being explicitly closed using Spring’s own JmsUtils. Turns out, the CachingConnectionFactory‘s JavaDoc held the key to what was going on: “Note also that MessageConsumers obtained from a cached Session won’t get closed until the Session will eventually be removed from the pool.” However, if the MessageConsumers could be reused this wouldn’t be an issue. Unfortunately, CachingConnectionFactory caches MessageConsumers based on a hash key which contains the selector among other values. Obviously each request/response call, with its necessarily unique selector, was creating a new consumer that could never be reused. Luckily ActiveMQ provides a PooledConnectionFactory which does not cache MessageConsumers and switching to it fixed the problem instantly. However, this means that each request/response requires a new MessageConsumer to be created. This is adds overhead but its the price that must be payed to do synchronous request/response.