Camel ActiveMQ性能调整

Question

情况

目前,我们在ActiveMQ库之上使用一些自定义代码来进行JMS消息传递.我一直在寻求切换到Camel,易于使用,易于维护和可靠性.

问题

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

测试方法

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

题

我是否可以根据发送到消息洪流的时间和个别消息的时间来改进下面的实现？理想情况下,改进将涉及调整我错过的一些配置,或建议更好的方法来做,而不是太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);

然后在你的路线中：

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

此外，在 ActiveMQ 中，您可以使用虚拟目的地。虚拟主题将充当队列，然后您可以使用与普通队列相同的 concurrentConsumers 方法。

进一步编辑（用于发送）：

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

编辑

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

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

Answer 2

我接手了原始发布者的工作，将其视为另一个任务的一部分，发现丢失消息的问题实际上是在 ActiveMQ 配置中。

我们设置了 sendFailIfNoSpace=true，如果我们发送的速度足够快以填充发布者缓存，则会导致消息被丢弃。通过使用policyEntry主题缓存大小，我可以改变消失的消息数量，并以与此类竞争条件预期的可靠性相同的程度。设置 sendFailIfNoSpace=false （默认），我可以拥有我喜欢的任何缓存大小，并且永远不会无法接收所有消息。

理论上，sendFailIfNoSpace 在删除消息时应该抛出 ResourceAllocationException，但这要么没有发生（！），要么以某种方式被忽略。同样有趣的是，尽管我们的自定义 JMS 包装器代码运行吞吐量测试的速度比 Camel 快，但并未遇到此问题。也许该代码速度更快，这意味着发布缓存被更快地清空，否则我们将在我尚未找到的连接代码中重写 sendFailIfNoSpace 。

关于速度问题，到目前为止，除了虚拟目的地之外，我们已经实现了这里提到的所有建议，但是具有 100K 消息的 Camel 版本测试在我的机器上仍然运行了 16 秒，而我们自己的包装器则为 10 秒。如上所述，我隐隐怀疑我们正在（隐式或以其他方式）覆盖包装器中某处的配置，但我怀疑是否有任何东西会导致 ActiveMQ 内的性能大幅提升。

gwithake 提到的虚拟目的地可能会加速这个特定的测试，但大多数时候对于我们的实际工作负载来说，这不是一个合适的解决方案。