concurrentConsumers 不是从一开始就立即创建的

Question

我在 Spring-Boot 应用程序中使用 Camel 从 AMQ-Queue 进行路由。来自该队列的消息将被发送到 REST-Web 服务。它已经在使用这个代码行了：

from("amq:queue:MyQueue").process("jmsToHttpProcessor").to(uri);

我的 uri 看起来像这样：

http4://localhost:28010/application/createCustomer

现在我要求到 Web 服务的路由应该并行完成：

为了实现这一点，我在 JmsConfiguration 中配置了 concurrentConsumers，如下所示：

    @Bean
    public JmsComponent amq(@Qualifier("amqConnectionFactory") ConnectionFactory amqConnectionFactory, AMQProperties amqProperties) {
        JmsConfiguration jmsConfiguration = new JmsConfiguration(amqConnectionFactory);
        jmsConfiguration.setConcurrentConsumers(50);
        jmsConfiguration.setMaxConcurrentConsumers(50);

        return new JmsComponent(jmsConfiguration);
    }

    @Bean
    public ConnectionFactory amqConnectionFactory(AMQProperties amqProperties) throws Exception {
        ConnectionFactoryParser parser = new ConnectionFactoryParser();
        ConnectionFactory returnValue = parser.newObject(parser.expandURI(amqProperties.getUrl()), "amqConnectionFactory");
        return returnValue;
    }

它按预期工作，但不是从一开始就立即。我有现象：

1. 我的 ActiveMQ 队列中有 100 条消息
2. 我启动我的 Spring 应用程序
3. Camel 在前一条消息得到响应后仅创建 1 个线程消耗 1 条消息
4. 我观察到队列中的消息数量只会缓慢减少（99.... 98... 97... 96...）
5. 我正在用新的 100 条消息填充队列
6. 现在正在创建并发消费者，因为我可以观察到消息迅速减少。

有人知道，为什么 concurrentConsumers 不能从一开始就立即工作吗？

Answer 1

我尝试了这些建议。不幸的是，他们没有改变行为。我发现，问题在于 Camel 在 Spring Boot 应用程序启动之前已经开始使用队列中的消息。我可以从日志中观察到这一点：

2021-04-01T20:26:33,901 INFO  (Camel (CamelBridgeContext) thread #592 - JmsConsumer[MyQueue]) [message]; ...
2021-04-01T20:26:33,902 INFO  (Camel (CamelBridgeContext) thread #592 - JmsConsumer[MyQueue]) [message]; ...
2021-04-01T20:26:33,915 INFO  (main) [AbstractConnector]; _; Started ServerConnector@5833f5cd{HTTP/1.1,[http/1.1]}{0.0.0.0:23500}
2021-04-01T20:26:33,920 INFO  (main) [BridgeWsApplication]; _; Started BridgeWsApplication in 12.53 seconds (JVM running for 13.429)

在这种情况下，只有一个线程 #592 的消费者正在消费所有消息。

实际上，如果我先启动我的 Spring 应用程序，然后用消息填充队列，那么将使用 concurrentConsumers：

2021-04-01T20:30:20,159 INFO  (Camel (CamelBridgeContext) thread #594 - JmsConsumer[MyQueue]) 
2021-04-01T20:30:20,159 INFO  (Camel (CamelBridgeContext) thread #599 - JmsConsumer[MyQueue]) 
2021-04-01T20:30:20,178 INFO  (Camel (CamelBridgeContext) thread #593 - JmsConsumer[MyQueue]) 
2021-04-01T20:30:20,204 INFO  (Camel (CamelBridgeContext) thread #564 - JmsConsumer[MyQueue]) 

在这种情况下，消息从 concurrentConsumers 并行消费。

为了解决问题，我尝试在我的 RouteBuilder 组件中将 autoStartUp 设置为 false：

    @Override
    public void configure() {

        CamelContext context = getContext();
        context.setAutoStartup(false);

        // My Route

    }

在我幼稚的想法中，我让Camel在Spring boot启动并运行后启动：

    public static void main(String[] args) {

        ConfigurableApplicationContext context = SpringApplication.run(BridgeWsApplication.class, args);

        SpringCamelContext camel = (SpringCamelContext) context.getBean("camelContext");
        camel.start();
        try {
            camel.startAllRoutes();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

不幸的是，这不会改变行为。 Spring启动后必须有配置让Camel启动。