入站通道适配器不根据执行程序缩放

Question

我有 jdbc:inbound-channel-adapter 轮询 50 条记录。我试图通过将 pollerExecutor 池大小缩放到 1-10 来提高性能，以便多个线程每个可以处理 50 条记录：

    <int-jdbc:inbound-channel-adapter
    id="initial.ContactType.poller"
    query="${poller.ContactType.get}"
    max-rows="${poller.deliveryContactType.maxRow:50}"
    row-mapper="ContactTypePollerRowMapper"
    data-source="dataSource" channel="ContactTypeChannel">
    <int:poller  fixed-rate="3000" time-unit="MILLISECONDS" task-executor="pollerExecutor">
        <int:advice-chain>
            <ref bean="pollerLoggingAdvice"/>
            <ref bean="txAdvice"  />
        </int:advice-chain>
    </int:poller>
</int-jdbc:inbound-channel-adapter>
<task:executor id="pollerExecutor" pool-size="1-10"
    queue-capacity="0" rejection-policy="CALLER_RUNS" />

我测试了无论池大小如何，处理 100,000 条记录所花费的时间都是相同的。

我分别用pool-size=1、pool-size=1-3和pool-size=1-10做了三轮测试，在所有三个测试中，每次 100,000 条记录花费 1 小时。

我通过检查日志确认 pollerExecutor 线程没有并行工作。 pollerExecutor-1 在 pollerExecutor-2 开始处理之前处理所有 50 条记录。

为什么 container/pollerExecutor 不能并行工作？

Answer 1

我认为你的问题出在这里：

/**
 * Set the capacity for the ThreadPoolExecutor's BlockingQueue.
 * Default is {@code Integer.MAX_VALUE}.
 * <p>Any positive value will lead to a LinkedBlockingQueue instance;
 * any other value will lead to a SynchronousQueue instance.
 * @see java.util.concurrent.LinkedBlockingQueue
 * @see java.util.concurrent.SynchronousQueue
 */
public void setQueueCapacity(int queueCapacity) {

因此，如果您指定 queue-capacity="0"，那么您最终会得到一个 SynchronousQueue，它无法接受新的并行任务，因为已经有一个忙于处理这 50 条记录。

尝试有一些合理的queue-capacity观察可能的并行性。

入站通道适配器不根据执行程序缩放

inbound-channel-adapter not scaling based on executor

spring-integration

executor

threadpoolexecutor