如何使Spring JMSListener突发到最大并发线程数？

Question

我有一个使用ActiveMQ5.10版的Spring JMS应用程序。我正在执行一个简单的并发测试。我正在使用Spring Boot、当前版本和注释来配置JMSListener和消息生产者。

消息生成器只是尽可能快地在队列中抛出消息。消息侦听器将消息从队列中取出，但在获得消息后会Hibernate1秒--模拟消息侦听器在获得消息后需要做的一些工作。

我将JMSListener设置为100-1000个并发线程。如果我同时启动消息生产者和消费者（两者都运行在各自的JVM中），那么消费者永远不会超过最小配置线程，即使最大范围设置为1000。

如果我让生产者首先启动并在队列中放置几千条消息，然后启动消费者的一个或多个实例，它将稳定地增加线程，从每秒100个线程开始，然后20个线程左右，直到队列中有大约20-30条正在运行的消息。它从不捕获生产者--队列中总是有一些消息，即使消费者不接近它的maxConcurrency计数。

@Component
public class ClientServiceImpl implements ClientService {

    private static final String QUEUE="message.test.queue";

    @Autowired
    private JmsTemplate jmsTemplate;

    @Override
    public void submitMessage(ImportantMessage importantMessage) {

        System.out.println("*** Sending " + importantMessage);
        jmsTemplate.convertAndSend(QUEUE, importantMessage);

    }
}

@SpringBootApplication
@EnableJms
public class AmqConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(AmqConsumerApplication.class, args);
    }
    @Value("${JMSHost}")
    private String JMS_BROKER_URL;

    @Autowired
    static Command command;

    @Bean
    public ConnectionFactory connectionFactory() {
        ConnectionFactory factory= new ActiveMQConnectionFactory(JMS_BROKER_URL);
        ((ActiveMQConnectionFactory)factory).setTrustAllPackages(true);
        ((ActiveMQConnectionFactory)factory).setOptimizeAcknowledge(true);
        ((ActiveMQConnectionFactory)factory).setAlwaysSessionAsync(false);
        return factory;
    }

}

@Component
public class TransformationListener {

    private static final String QUEUE="message.test.queue?consumer.prefetchSize=10";

    @JmsListener(destination=QUEUE, concurrency = "100-1000")
    public void handleRequest(ImportantMessage importantMessage) {
        System.out.println("*** Recieved message: " + importantMessage + " on thread" + Thread.currentThread().getId());
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}

Answer 1

你还在面对这种行为吗？您是否在http://activemq.apache.org/what-is-the-prefetch-limit-for.html上阅读了这篇“pooled consumer and prefetch”的建议，您是否尝试使用prefetchsize=0或1？我想1能解决你的问题。如果prefetchSize大于1，可能需要将AbortSlowAckConsumerStrategy降低到低于默认的30s。在您的情况下，要有超过100个线程消耗消息，需要队列中超过1000个未消耗和未预取的消息，因为prefetchSize是10。