Camel拆分器并行处理阵列列表-并发访问问题

Question

使用Camel拆分数组列表，并在多达10个线程中并行处理每个项目。以下是配置。html" target="_blank">线程池配置文件被设置为最大线程数=10。

<camel:route id="ReportsRoute">
        <camel:from uri="direct:processReportsChannel" />
        <camel:to uri="bean:reportRepository?method=getPendingTransactions" />
        <camel:split parallelProcessing="true" executorServiceRef="ReportThreadPoolProfile">
            <camel:simple>${body}</camel:simple>
            <camel:doTry>
                <camel:to uri="direct:processReportChannel" />
                <camel:doCatch>
                    <camel:exception>java.lang.Exception</camel:exception>
                    <camel:handled>
                        <camel:constant>true</camel:constant>
                    </camel:handled>                        
                    <camel:to uri="bean:ReportRepository?method=markAsFailed"/>
                    <camel:wireTap uri="direct:loggingAndNotificationChannel" />
                </camel:doCatch>
            </camel:doTry>
        </camel:split>
    </camel:route>  

bean:reportRepository？method=getPendingTransactions获取ArrayList并传递给Splitter。

进程报告通道是处理项目的处理器。

问题:当作业开始时，它正在启动10个线程，但是一些线程正在拾取同一个项目。例如，如果我在数组列表中有item_no_1到10，thread_no_1和thread_no_2或者更多的线程正在获取，比如item_no_2。是因为数组列表不是线程安全的，而Splitter没有管理它吗？

我不是这方面的专家，需要帮助来指出问题在哪里。

Answer 1

我使用以下（更简单的）设置进行了测试：

<camelContext xmlns="http://camel.apache.org/schema/spring">
    <route id="ReportsRoute">
        <from uri="direct:start" />
         <!-- By default a pool size of 10 is used. -->
        <split parallelProcessing="true">
            <simple>${body}</simple>
            <to uri="direct:sub" />
        </split>
    </route>  
    <route>
        <from uri="direct:sub"/>
        <log message="Processing item ${body}" />
    </route>
</camelContext>

测试：

 List<Object> list = new ArrayList<>();
 for (int i = 0; i < 1000; i++) {
    list.add("And we go and go: " + (i + 1));
 }
 template.sendBody("direct:start", list);

使用此设置时，不会对任何条目进行两次处理。因此，处理器中一定存在导致问题的内容，即同一列表项被多个线程选取。