情况
目前，我们在JMS消息传递中使用一些自定义代码，基于ActiveMQ库。我一直在考虑切换到Camel，因为它易于使用、易于维护和可靠。
问题
在我的现有配置中，Camel的ActiveMQ实现比我们的旧实现慢得多，无论是每个消息发送和接收的延迟时间，还是发送和接收大量消息所需的时间。我已经尝试过调整一些配置（例如最大连接），但没有效果。
测试方法
我有两个应用程序，一个使用我们的旧实现，一个使用Camel实现。每个应用程序都向本地ActiveMQ服务器上的主题发送JMS消息，并侦听该主题上的消息。这用于测试两种情况： - 在循环中将100,000条消息发送到主题，并查看从开始发送到处理完所有消息所需的时间。 - 每100毫秒发送一条消息，并测量从发送到处理每条消息的延迟时间（以纳秒为单位）。
问题
我能否改进以下实现，以提高洪水消息和单个消息的发送时间和处理时间？理想情况下，改进将涉及调整我错过的某些配置或建议更好的方法，并且不会太过于hacky。欢迎解释改进。
编辑：现在我正在异步发送消息，似乎存在并发问题。receivedCount未达到100,000。查看ActiveMQ Web界面，有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();
}
}
}
threads(10)似乎让它稍微变差了...我还没有尝试虚拟目标方法。我会回复您这个方法的效果如何。实际上，我认为限制在发送端，因为它似乎在完成发送的同时也完成了处理。您有什么想法可以加快发送速度吗？ - Spycho