在不阻塞队列中的其他消息的情况下，将Spring JMSTemplate/ActiveMQ与请求/响应一起使用

Question

我们为我们想要等待响应的特定用例实现了分布式请求/响应类型体系结构。我们使用的JMS代理是ActiveMq，代码使用Spring连接在一起。

我们看到的问题是，如果向同一个目的地发送一组请求，那么任何需要花费大量时间才能完成的请求都会阻塞随后的请求消息。使用者使用的SessionAwareMessageListener接口只支持onMessage（）方法。在这里实现并行性的最佳方法是什么？即，如果某个特定请求需要很长时间，则不应阻止队列中的其他消息？

有这么一篇帖子，但它没有回答我的问题。JMS：我们可以通过提交或回滚从OnMessage（）中的队列中获取多条消息吗

谢啦

相关代码片段（为了简洁起见，删除了异常处理等）

制作人

public class MyJmsProducer {

private ProcessingResponse sendMessage(final Serializable serializable) {
    //send JMS request and wait for response
    return jmsMessagingTemplate.convertSendAndReceive(destination, serializable, ProcessingResponse.class); //this operation seems to be blocking + sync
   }
}

而听者（消费者）

public class MyJmsListener
    implements SessionAwareMessageListener<Message>, NotificationHandler<Task> {

@Override
public void onMessage(Message message, Session session)
        throws JMSException {
    ProcessingRequest processingRequest = (ProcessingRequest) ((ObjectMessage) message).getObject();

    // handle the request here (THIS COULD TAKE A WHILE)
    handleRequest(processingRequest);


    // done handling the request, now create a response message
    final ObjectMessage responseMessage = new ActiveMQObjectMessage();
    responseMessage.setJMSCorrelationID(message.getJMSCorrelationID());
    responseMessage.setObject(processingResponse);

    // Message sent back to the replyTo address of the income message.
    final MessageProducer producer = session.createProducer(message.getJMSReplyTo());
    producer.send(responseMessage);

  }
}

Answer 1

您可以使用DMLC的Conextt消费者来提高消息的消耗速度并解决消耗慢的问题：

@Bean
public DefaultMessageListenerContainer dmlc() {
    DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
    dmlc.setMaxConcurrentConsumers(10);
    dmlc.setConcurrentConsumers(5);
    return dmlc;
}

您需要调整预取策略以适应并发消费者：

persistent queues (default value: 1000)
non-persistent queues (default value: 1000)
persistent topics (default value: 100)
non-persistent topics (default value: Short.MAX_VALUE - 1)

所有消息都被发送到第一个连接的消费者，当另一个消费者连接到同一个目的地时，他不会收到消息，因此要更改此行为，需要将prefetchPolicy设置为低于默认值的值。例如，添加这个jms。预取策略。queuePrefetch=1到activemq中的uri配置。xml或在客户端url上设置它，如下所示

@Bean
public ConnectionFactory connectionFactory() {
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
            "tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=1");
    return connectionFactory;
}

对于高消息量的高性能，建议使用大的预取值。但是，对于较低的消息量，其中每个消息需要很长时间来处理，预取应该设置为1。这可以确保消费者一次只处理一条消息。但是，将预取限制指定为零将导致消费者轮询消息，一次轮询一个消息，而不是将消息推送给消费者。

看看http://activemq.apache.org/what-is-the-prefetch-limit-for.html

和

http://activemq.apache.org/destination-options.html