如何控制spring远程批处理从机

Question

<bean id="ourTaskExecutor" class="...ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="16" />
    <property name="maxPoolSize" value="16" />
    <property name="queueCapacity" value="256" />
</bean>

<int-jms:inbound-channel-adapter 
    id="JobJmsRequests" 
    connection-factory="jmsConnectionFactory"
    channel="requests.chunking" 
    destination-name="requests.chunking" >
        <int:poller task-executor="ourTaskExecutor" 
            fixed-delay="50" 
            max-messages-per-poll="4" 
            receive-timeout="50"/>
</int-jms:inbound-channel-adapter>

<int-jms:outbound-channel-adapter 
    id="JobJmsReplies" 
    connection-factory="jmsConnectionFactory" 
    destination-name="replies.chunking" 
    channel="replies.chunking"/>

<int:service-activator 
    id="JobActivator" 
    input-channel="requests.chunking" 
    output-channel="replies.chunking" 
    ref="JobChunkHandler" method="handleChunk"/> 

我们的假设是poller会将接收到的消息交给task executor进行处理，但是task executor中的消息不会超过64条，而task executor可以容纳256+16条消息。

然而，我们看到异常，队列容量已经达到，任务执行器拒绝任务。我们认为这是因为任务执行器仅由轮询器使用，并在轮询线程上处理获得的请求。由于处理需要很长时间，我们慢慢地用轮询任务淹没任务执行器。并在单个线程中处理它们接收到的所有内容）。

我们不能完全解释泛洪，因为我们使用了receivetimeout和延迟，所以我们的理解是，应该在触发新的轮询之前进行处理。然而，情况似乎并非如此。

我们希望在不同的线程上轮询和执行消息。这可能吗？如何实现？

Answer 1

使用caller-runs或caller-blocksRejectedExecutionHandler。

也就是说，您不想要内存中的块--它们将在系统故障时丢失。

应使用消息驱动的入站通道适配器，其并发性设置为要处理的并发块数。