Apache Camel:asyncCallback回复线程在Camel 17和Camel 18/19中更改

Question

对于高达17.x的Camel版本,Synchronization.onComplete()回调上的所有调用activemq:queue都是在不同的线程中完成的,因此即使onComplete中的响应处理非常慢,消息也没有被阻塞和排队.asyncConsumer=true&defaultTaskExecutorType=ThreadPool&concurrentConsumers=2&maxConcurrentConsumers=100根据我的理解,这应该是配置的结果.因此,此展示的示例输出是:

Received async reply: 2000 OK
Received async reply: 3000 OK
Received async reply: 5000 OK
Received async reply: 9000 OK
Received async reply: 10000 OK
Finished async reply: 2000 OK
Finished async reply: 3000 OK
Finished async reply: 5000 OK
Finished async reply: 9000 OK
Finished async reply: 10000 OK

所以"完成"日志毕竟是"收到",因为每个日志都在另一个线程中调用.所有回复都是异步接收的,处理的时间长短不会影响其他回复的接收.

升级到Camel 18.x(或19.x)后,这已不再相同.现在收到回复和处理(漫长的过程)它阻止接收其他人.这是因为相同的线程用于调用Synchroniztion.onComplete(),因此回复排队直到完全处理.

Received async reply: 2000 OK
Received async reply: 10000 OK
Received async reply: 9000 OK
Finished async reply: 2000 OK
Received async reply: 3000 OK
Finished async reply: 3000 OK
Finished async reply: 9000 OK
Finished async reply: 10000 OK
Received async reply: 5000 OK
Finished async reply: 5000 OK

我认为这些新属性的配置完全如下:replyToConcurrentConsumers = 2&replyToMaxConcurrentConsumers = 100并且如果有新消息,如果所有当前线程都正在处理(当然如果未达到最大计数),它将在新线程中处理和回复那么处理多条消息并排队的线程是正常的)

也许我做错了但是如何配置路由以便获得类似于Camel 17的结果.增加replyToConcurrentConsumers属性它可以工作,但我认为它应该动态扩展并依赖于replyToMaxConcurrentConsumers,如果需要的话.

代码 :

public class AsyncCallbackTest {

private static final Logger LOG = LoggerFactory.getLogger(AsyncCallbackTest.class);

public static void main(String[] args) throws Exception {
    // create and setup a default Camel context
    final CamelContext camelContext = new DefaultCamelContext();
    setupRoutes(camelContext);
    camelContext.start();

    // real test
    asyncCallbackTest(camelContext);

    camelContext.stop();
    System.exit(0);
}

private static void setupRoutes(CamelContext camelContext) throws Exception {
    camelContext.addRoutes(new RouteBuilder() {
        public void configure() {
            from("activemq://queue:asyncTest?asyncConsumer=true&defaultTaskExecutorType=ThreadPool" +
                    "&concurrentConsumers=2&maxConcurrentConsumers=100")
                    .process(exchange -> {
                        final String msg = String.valueOf(exchange.getIn().getBody());
                        exchange.getOut().setBody(msg + " OK");
                    });
        }
    });
}

private static void asyncCallbackTest(CamelContext camelContext) throws Exception {
    final int[] delays = new int[]{9000, 10000, 5000, 2000, 3000};

    final CountDownLatch countDownLatch = new CountDownLatch(delays.length);

    final Synchronization callback = new SynchronizationAdapter() {
        @Override
        public void onComplete(Exchange exchange) {
            LOG.info("Received async reply: " + exchange.getOut().getBody());
            final int delay = (int) exchange.getIn().getBody();
            synchronized (this) {
                try {
                    this.wait(delay);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            LOG.info("Finished async reply: " + exchange.getOut().getBody());

            super.onComplete(exchange);
        }

        @Override
        public void onDone(Exchange exchange) {
            countDownLatch.countDown();
        }
    };

    final ProducerTemplate producerTemplate = camelContext.createProducerTemplate();

    for (int i = 0; i < delays.length; i++) {
        final Exchange exchange = new DefaultExchange(camelContext);
        exchange.getIn().setBody(delays[i]);
        exchange.setPattern(ExchangePattern.InOut);

        producerTemplate.asyncCallback("activemq://queue:asyncTest?" +
                        "replyToConcurrentConsumers=2&replyToMaxConcurrentConsumers=100",
                exchange, callback);
    }

    countDownLatch.await();
    camelContext.stop();

}

}