java - Camel ActiveMQ 性能调优

Question

情况

目前，我们在 ActiveMQ 库之上使用一些自定义代码来进行 JMS 消息传递。我一直在考虑切换到 Camel，以便于使用、维护和可靠性。

问题

使用我目前的配置，Camel 的 ActiveMQ 实现比我们的旧实现要慢得多，无论是在发送和接收每条消息的延迟方面，还是在发送和接收大量消息所花费的时间方面。我试过调整一些配置(例如最大连接数)，但无济于事。

测试方法

我有两个应用程序，一个使用我们的旧实现，一个使用 Camel 实现。每个应用程序将 JMS 消息发送到本地 ActiveMQ 服务器上的一个主题，并且还监听关于该主题的消息。这用于测试两个场景:- 在循环中向主题发送 100,000 条消息，并查看从开始发送到处理完所有消息需要多长时间。- 每 100 毫秒发送一条消息并测量从发送到处理每条消息的延迟(以纳秒为单位)。

问题

我能否改进下面的实现，在发送时间到处理消息泛滥和单个消息的时间方面？理想情况下，改进将包括调整一些我错过的配置，或者建议一个更好的方法来做，而不是太 hacky。对改进的解释将不胜感激。

编辑:既然我正在异步发送消息，我似乎遇到了并发问题。 receivedCount 未达到 100,000。查看ActiveMQ web界面，100,000条消息入队，100,000条出队，所以这可能是消息处理端的问题。我已将 receivedCount 更改为 AtomicInteger 并添加了一些日志记录以帮助调试。这可能是 Camel 本身(或 ActiveMQ 组件)的问题，还是消息处理代码有问题？据我所知，只有大约 99,876 条消息通过 floodProcessor.process。

测试实现

编辑:针对并发问题更新了异步发送和日志记录。

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsConfiguration;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.log4j.Logger;

public class CamelJmsTest{
    private static final Logger logger = Logger.getLogger(CamelJmsTest.class);

    private static final boolean flood = true;
    private static final int NUM_MESSAGES = 100000;

    private final CamelContext context;
    private final ProducerTemplate producerTemplate;

    private long timeSent = 0;

    private final AtomicInteger sendCount = new AtomicInteger(0);
    private final AtomicInteger receivedCount = new AtomicInteger(0);

    public CamelJmsTest() throws Exception {
        context = new DefaultCamelContext();

        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

        PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(connectionFactory);

        JmsConfiguration jmsConfiguration = new JmsConfiguration(pooledConnectionFactory);
        logger.info(jmsConfiguration.isTransacted());

        ActiveMQComponent activeMQComponent = ActiveMQComponent.activeMQComponent();
        activeMQComponent.setConfiguration(jmsConfiguration);

        context.addComponent("activemq", activeMQComponent);

        RouteBuilder builder = new RouteBuilder() {
            @Override
            public void configure() {
                Processor floodProcessor = new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        int newCount = receivedCount.incrementAndGet();

                        //TODO: Why doesn't newCount hit 100,000? Remove this logging once fixed
                        logger.info(newCount + ":" + exchange.getIn().getBody());

                        if(newCount == NUM_MESSAGES){
                            logger.info("all messages received at " + System.currentTimeMillis());
                        }
                    }
                };

                Processor spamProcessor = new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        long delay = System.nanoTime() - timeSent;

                        logger.info("Message received: " + exchange.getIn().getBody(List.class) + " delay: " + delay);
                    }
                };

                from("activemq:topic:test?exchangePattern=InOnly")//.threads(8) // Having 8 threads processing appears to make things marginally worse
                    .choice()
                        .when(body().isInstanceOf(List.class)).process(flood ? floodProcessor : spamProcessor)
                    .otherwise().process(new Processor() {
                        @Override
                        public void process(Exchange exchange) throws Exception {
                            logger.info("Unknown message type received: " + exchange.getIn().getBody());
                        }
                    });
            }
        };

        context.addRoutes(builder);

        producerTemplate = context.createProducerTemplate();
        // For some reason, producerTemplate.asyncSendBody requires an Endpoint to be passed in, so the below is redundant:
//      producerTemplate.setDefaultEndpointUri("activemq:topic:test?exchangePattern=InOnly");
    }

    public void send(){
        int newCount = sendCount.incrementAndGet();
        producerTemplate.asyncSendBody("activemq:topic:test?exchangePattern=InOnly", Arrays.asList(newCount));
    }

    public void spam(){
        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                timeSent = System.nanoTime();
                send();
            }
        }, 1000, 100, TimeUnit.MILLISECONDS);
    }

    public void flood(){
        logger.info("starting flood at " + System.currentTimeMillis());
        for (int i = 0; i < NUM_MESSAGES; i++) {
            send();
        }
        logger.info("flooded at " + System.currentTimeMillis());
    }

    public static void main(String... args) throws Exception {
        CamelJmsTest camelJmsTest = new CamelJmsTest();
        camelJmsTest.context.start();

        if(flood){
            camelJmsTest.flood();
        }else{
            camelJmsTest.spam();
        }
    }
}

Answer 1

从您当前的 JmsConfiguration 看来，您只使用单个线程使用消息。这是故意的吗？

如果不是，您需要将 concurrentConsumers 属性设置为更高的值。这将创建一个 JMS 监听器线程池来为您的目的地提供服务。

例子:

JmsConfiguration config = new JmsConfiguration(pooledConnectionFactory);
config.setConcurrentConsumers(10);

这将创建 10 个 JMS 监听器线程，它们将并发处理来自您的队列的消息。

编辑:

对于主题你可以这样做:

JmsConfiguration config = new JmsConfiguration(pooledConnectionFactory);
config.setConcurrentConsumers(1);
config.setMaxConcurrentConsumers(1);

然后在你的 route :

from("activemq:topic:test?exchangePattern=InOnly").threads(10)

此外，在 ActiveMQ 中，您可以使用 virtual destination .虚拟主题将像队列一样工作，然后您可以使用与普通队列相同的 concurrentConsumers 方法。

进一步编辑(用于发送):

您当前正在进行阻塞发送。您需要执行 producerTemplate.asyncSendBody()。

---

编辑

我刚刚用您的代码构建了一个项目并运行了它。我在您的 floodProcessor 方法中设置了一个断点，newCount 达到了 100,000。我认为您可能会被您的日志记录以及您异步发送和接收的事实所困扰。在我的机器上，newCount 达到了 100,000，"all messages received" 消息在执行后不到 1 秒就被记录了下来，但是程序在缓冲后又继续记录了 45 秒。通过减少日志记录，您可以看到日志记录对您的 newCount 数字与 body 数字的接近程度的影响。我将日志记录转为 info，关闭了 camel 日志记录，并且在日志记录末尾匹配了两个数字:

INFO  CamelJmsTest - 99996:[99996]
INFO  CamelJmsTest - 99997:[99997]
INFO  CamelJmsTest - 99998:[99998]
INFO  CamelJmsTest - 99999:[99999]
INFO  CamelJmsTest - 100000:[100000]
INFO  CamelJmsTest - all messages received at 1358778578422