我需要aggregate方法在我的split中并行运行，所以我激活了parallelAggregate选项，但它没有像我预期的那样工作。
@RunWith(SpringJUnit4ClassRunner.class)
public class SplitAggregationTest extends CamelTestSupport {
private final Logger LOGGER = LoggerFactory.getLogger(SplitAggregationTest.class);
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
from("direct:start")
.split(body(), new AggregationStrategy() {
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
LOGGER.error("InterruptedException: ", e);
}
return oldExchange == null ? newExchange : oldExchange;
}
}).streaming().parallelProcessing().parallelAggregate()
.log(LoggingLevel.INFO, LOGGER, "Aggreg ${body}")
.end();
}
};
}
@Test
public void test1() throws InterruptedException {
Thread.sleep(5000);
template.sendBody("direct:start", Arrays.asList("A", "B", "C", "D"));
Thread.sleep(5000);
}聚合方法（紫色）不是并行运行的，它们会在以下时间之前等待对方完成：
编辑-相应的线程转储：
"Camel (camel-2) thread #17 - Split" #36 daemon prio=5 os_prio=0 tid=0x000002317da51800 nid=0x149bc waiting on condition [0x00000062a55fe000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x000000076e105c70> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:870)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1199)
at java.util.concurrent.locks.ReentrantLock$NonfairSync.lock(ReentrantLock.java:209)
at java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:285)
at org.apache.camel.util.concurrent.AsyncCompletionService.complete(AsyncCompletionService.java:141)
at org.apache.camel.util.concurrent.AsyncCompletionService.access$200(AsyncCompletionService.java:30)
at org.apache.camel.util.concurrent.AsyncCompletionService$Task.accept(AsyncCompletionService.java:168)
at org.apache.camel.processor.MulticastProcessor$MulticastReactiveTask.lambda$null$0(MulticastProcessor.java:579)
at org.apache.camel.processor.MulticastProcessor$MulticastReactiveTask$$Lambda$519/1049021104.done(Unknown Source)
at org.apache.camel.AsyncCallback.run(AsyncCallback.java:44)
at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:187)
at org.apache.camel.impl.engine.DefaultReactiveExecutor.schedule(DefaultReactiveExecutor.java:59)
at org.apache.camel.processor.MulticastProcessor.lambda$schedule$1(MulticastProcessor.java:348)
at org.apache.camel.processor.MulticastProcessor$$Lambda$517/1626228981.run(Unknown Source)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Locked ownable synchronizers:
- <0x000000076f0ca838> (a java.util.concurrent.ThreadPoolExecutor$Worker)
"Camel (camel-2) thread #16 - Split" #35 daemon prio=5 os_prio=0 tid=0x000002317da54000 nid=0x6a8 waiting on condition [0x00000062a54fe000]
java.lang.Thread.State: TIMED_WAITING (sleeping)
at java.lang.Thread.sleep(Native Method)
at com.mybatch.camel.processor.SplitAggregationTest$1$1.aggregate(SplitAggregationTest.java:34)
at org.apache.camel.AggregationStrategy.aggregate(AggregationStrategy.java:86)
at org.apache.camel.processor.MulticastProcessor.doAggregateInternal(MulticastProcessor.java:894)
at org.apache.camel.processor.MulticastProcessor.doAggregate(MulticastProcessor.java:858)
at org.apache.camel.processor.MulticastProcessor$MulticastTask.aggregate(MulticastProcessor.java:451)
at org.apache.camel.processor.MulticastProcessor$MulticastReactiveTask.lambda$null$0(MulticastProcessor.java:582)
at org.apache.camel.processor.MulticastProcessor$MulticastReactiveTask$$Lambda$519/1049021104.done(Unknown Source)
at org.apache.camel.AsyncCallback.run(AsyncCallback.java:44)
at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:187)
at org.apache.camel.impl.engine.DefaultReactiveExecutor.schedule(DefaultReactiveExecutor.java:59)
at org.apache.camel.processor.MulticastProcessor.lambda$schedule$1(MulticastProcessor.java:348)
at org.apache.camel.processor.MulticastProcessor$$Lambda$517/1626228981.run(Unknown Source)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Locked ownable synchronizers:
- <0x000000076e105c70> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)
- <0x000000076ef115b0> (a java.util.concurrent.ThreadPoolExecutor$Worker)
2条答案
按热度按时间e0bqpujr1#
让我们首先定义这些术语：
并行处理：如果为true，则每个拆分项都将通过线程池（默认或您提供的线程池）中的线程运行。
并行聚合：如果为true，Camel将不同步对
AggregationStrategy的调用（顺便说一句，确保它是线程安全的）。根据您获得的线程名称，Camel正在不同的线程上运行拆分项。您确定您的
aggregate方法没有被多个线程并发调用吗？您可能看到Camel等待所有拆分项完成，然后才越过拆分。i34xakig2#
经过一些调查（src here），MulticastProcessor#agregate（）中的reentrantlock似乎阻止了拆分线程同时处理camel V3.X中的
AggregationStrategy（在V2.X中，与//中预期的测试聚合相同）你不认为这种聚合方法中的reentrantlock不是并行聚合配置的意义吗？
以下是测试的日志输出（在1秒内依次进行聚合处理）
这里是“pool-1-thread-1”聚集时的线程转储。其他3个线程正在等待停止以等待 ReentrantLock$NonfairSync
谢谢你的帮助