如何配置 Spring Integration 以使用多线程从 Redis 读取？

【问题标题】：how to configure Spring Integration to read from Redis with multithread?如何配置 Spring Integration 以使用多线程从 Redis 读取？
【发布时间】：2019-10-16 01:44:20
【问题描述】：

我正在设置一个 Spring Integration 配置以使用多线程从 redis 读取，但是当我运行我的应用程序时，Spring 只创建一个线程。

我正在创建一个 int-redis:queue-inbound-channel-adapter，其中的 executor-task 具有 pool-size= 500 和 queue-capacity=0。

<redis:queue-inbound-channel-adapter
            id="fromRedis" channel="privateAggregationExecutorChannel" queue="${instance}_private"
            receive-timeout="1000" recovery-interval="3000" expect-message="false" error-channel="distributionErrors"
            auto-startup="false" task-executor="robotTaskExecutor"/>

<task:executor
            id="robotTaskExecutor"
            pool-size="500"
            queue-capacity="0"
            keep-alive="50"
            rejection-policy="CALLER_RUNS" />

<int:service-activator input-channel="privateAggregationExecutorChannel" ref="aggregationExecutor" method="run" />

我不知道我做错了什么，或者我是否缺少某些东西。感谢您的帮助。

【问题讨论】：

标签： spring multithreading redis spring-integration

【解决方案1】：

没错。 RedisQueueMessageDrivenEndpoint 是真正的单线程组件：

@Override
protected void doStart() {
    if (!this.active) {
        this.active = true;
        this.restart();
    }
}

private void restart() {
    this.taskExecutor.execute(new ListenerTask());
}

如您所见，只有一个ListenerTask 是从此通道适配器安排的。

要使其成为多线程，最好有一个ExecutorChannel 来从这个通道适配器发送消息。这样，即使RedisQueueMessageDrivenEndpoint 是单线程的，您仍将进行多线程处理。

我们刚刚意识到，当我们可以解决其他简单的方法时，将并发性引入这个组件会有点复杂。

另一种方法是为相同的queue 和same 通道定义多个<redis:queue-inbound-channel-adapter> 来发送。

【讨论】：