情况
目前，我们在 ActiveMQ 库之上使用一些自定义代码来进行 JMS 消息传递。我一直在考虑改用 Camel，因为它易于使用、易于维护和可靠性。
问题
在我目前的配置下，Camel 的 ActiveMQ 实现比我们的旧实现慢得多，无论是在发送和接收每条消息的延迟方面，还是在发送和接收大量消息所花费的时间方面。我尝试调整一些配置（例如最大连接数），但无济于事。
测试方法
我有两个应用程序，一个使用我们的旧实现，一个使用 Camel 实现。每个应用程序都将 JMS 消息发送到本地 ActiveMQ 服务器上的主题，并侦听有关该主题的消息。这用于测试两个场景： - 在循环中向主题发送 100,000 条消息，并查看从开始发送到结束处理所有消息需要多长时间。- 每 100 毫秒发送一条消息，并测量从发送到处理每条消息的延迟（以 ns 为单位）。
问题
我可以改进下面的实现，就发送时间到处理大量消息和单个消息的时间而言？理想情况下，改进将涉及调整我错过的一些配置，或者提出更好的方法来做到这一点，而且不要太老套。改进的解释将不胜感激。
编辑：现在我异步发送消息，我似乎有一个并发问题。receivedCount没有达到 100,000。看ActiveMQ的web界面，10万条消息入队，10万条出队，所以大概是消息处理端的问题。我已更改receivedCount为 anAtomicInteger并添加了一些日志记录以帮助调试。这可能是 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();
}
}
}