Spring集成：是否可以让发送者等到任务执行器队列/线程空闲。答案

【问题标题】：Spring integration: Is it possible to make the sender wait till the task executor queue/thread is free.Spring集成：是否可以让发送者等到任务执行器队列/线程空闲。
【发布时间】：2018-06-15 13:55:39
【问题描述】：

是否可以让发送者等到任务执行器队列/线程空闲。

任务执行者如下：

<int:channel id="inputRequestChannel">
        <int:dispatcher task-executor="validationWorkers"/>
   </int:channel>
   <task:executor id="validationWorkers" pool-size="5" queue-capacity="10"  rejection-policy="CALLER_RUNS"/>

使用上述配置会丢失一些数据包，这些数据包会发送到 inputRequestChannel。

如何保证不丢包？

哪种拒绝政策合适？

【问题讨论】：

>How to ensure that the no packets are lost? 那是不可能的；一旦它们在执行者的队列中，如果服务器崩溃，它们就会丢失。您需要显示其余配置，以便有人提出有关避免消息丢失的建议。

标签： spring-integration

【解决方案1】：

您是否介意分享一些测试用例来重现并真正理解数据包丢失。不确定“数据包”本身是什么......

您可以考虑改用CallerBlocksPolicy：

/**
 * A {@link RejectedExecutionHandler} that blocks the caller until
 * the executor has room in its queue, or a timeout occurs (in which
 * case a {@link RejectedExecutionException} is thrown.
 *
 * @author Gary Russell
 * @since 3.0.3
 *
 */
public class CallerBlocksPolicy implements RejectedExecutionHandler {

【讨论】：

感谢 Artem Bilan。问题得到解决。似乎问题在于主应用程序正在退出，为了验证，我在队列通道中的消息被消耗并打印消息之前打印了收到的消息。
好。是不是意味着我们可以接受stackoverflow.com/help/someone-answers的答案？

【解决方案2】：

只是为了补充 Artem 的答案，这是我使用 CallerBlocksPolicy 的示例：


    @Bean(name = "inChannel")
    public MessageChannel inChannel() {
        return MessageChannels.direct().get();
    }


    @Bean(name = "asyncChannelsExecutor")
    public Executor asyncChannelsExecutor() {
        final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(3);
        executor.setQueueCapacity(2);
        executor.setThreadNamePrefix("TTT");
        executor.setRejectedExecutionHandler(new CallerBlocksPolicy(30_000));
        executor.initialize();
        return executor;
    }

    @Bean(name = "asyncOutChannel")
    public MessageChannel asyncOutChannel() {
        return MessageChannels.executor(asyncChannelsExecutor()).get();
    }

    @Bean
    public IntegrationFlow asyncFlow() {
        return IntegrationFlows.from("inChannel")
            .transform(dtoTransformer)
            .channel("asyncOutChannel")
            .handle(handler)
            .get();
     }

注意org.springframework.integration.util.CallerBlocksPolicy 由 Spring Integration 提供。 java.util.concurrent.RejectedExecutionHandler 是核心 Java 平台接口。

【讨论】：