Camel ActiveMQ 性能调优

Posted 2023-03-10

【中文标题】Camel ActiveMQ 性能调优

【英文标题】：Camel ActiveMQ Performance Tuning

【发布时间】：2013-01-02 07:45:31

【问题描述】：

情况

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

问题

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

测试方法

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

问题

我可以改进下面的实现，就发送时间到处理大量消息和单个消息的时间而言？理想情况下，改进将涉及调整我错过的一些配置，或者提出更好的方法来做到这一点，而且不要太老套。改进的解释将不胜感激。

编辑：现在我正在异步发送消息，我似乎遇到了并发问题。 receivedCount 未达到 100,000。看ActiveMQ的web界面，10万条消息入队，10万条出队，所以大概是消息处理端的问题。我已将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();
        
    

【参考方案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 中，您可以使用 virtual destination。虚拟主题将像队列一样工作，然后您可以使用与普通队列相同的 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

【讨论】：

我确实尝试过，但是该设置实际上使每条消息被消耗 10 次。文档说它用于队列，而不是主题。有没有办法拥有一个消费者线程池，但只将每条消息发送给其中一个？

使用可在任何 JMS 服务器上工作的通用骆驼方法以及 ActiveMQ 实现特定方法进行编辑。

感谢您的编辑。然而，threads(10) 似乎让它变得更糟......我还没有尝试过虚拟目的地方法。我会回帖说明情况如何。我实际上认为限制在发送端，因为它似乎在完成发送的同一毫秒内完成处理。关于如何加快发送速度的任何想法？

使用异步发送调用编辑。

您现在是否看到有关该主题的消息？如果不是，那么您仍然受到发送的瓶颈，您也应该线程发送。在以前的项目中，我构建了 Camel 应用程序，每天处理数百万条 JMS 消息，以便 camel 可以处理它。这些使用 TibcoEMS 作为代理，但我认为您可以从 ActiveMQ 中获得类似的性能。

【参考方案2】：

我从最初的发帖人那里接手，将其视为另一项任务的一部分，发现丢失消息的问题实际上出在 ActiveMQ 配置中。

我们有一个设置 sendFailIfNoSpace=true，如果我们发送的速度足够快以填满发布者缓存，这会导致消息被丢弃。玩弄 policyEntry 主题缓存大小，我可以改变消失的消息数量，其可靠性与这种竞争条件的预期一样高。设置 sendFailIfNoSpace=false（默认），我可以拥有任何我喜欢的缓存大小，并且永远不会接收到所有消息。

理论上，sendFailIfNoSpace 在丢弃消息时应该抛出 ResourceAllocationException，但这要么没有发生（！），要么以某种方式被忽略。同样有趣的是，尽管运行吞吐量测试比 Camel 更快，但我们的自定义 JMS 包装器代码并没有遇到这个问题。也许该代码更快，这意味着发布缓存被更快地清空，或者我们在连接代码中我还没有找到的地方覆盖了 sendFailIfNoSpace。

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

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