java - Spring集成DSL : how to make single-threaded caching channel?

Question

我有以下流程配置:

@Bean
public IntegrationFlow setupDatabaseFlow() {
    return IntegrationFlows.from(pubSubDatabaseRabbitOutput)
          // WHAT CHANNEL
            .handle((p, h) -> databaseActivator.recordToDatabase(p, h))
            .get();
}

并且我愿意在“WHAT CHANNEL”处插入一些 channel ，即

• 缓存
• 单线程
• 在单独的线程中运行

此处使用的最佳 channel 规范是什么？

Answer 1

不确定用例是什么，但这可以通过 QueueChannel 和 PollingConsumer 以及 fixedDelay 策略来实现。

1. QueueChannel 使用队列缓冲区(持久的或内存中的)来保存消息，直到消息被使用为止。按照您的说法，A-la 缓存。

2. PollingConsumer 使用 TaskScheduler 从队列中进行轮询，因此是单独的线程。

3. 使用 fixedDelay 策略可以确保下一个轮询任务在前一个轮询任务完成之前不会启动。因此采用单线程模型。

如果您不喜欢在轮询任务之间 hibernate ，并最终采用事件驱动模型，则可以使用 fixedDelay(-1)，这样下一个轮询任务将不会执行上一次之后不要 sleep 。当前轮询任务确实会阻塞线程，直到队列中的值为止。

更新

关于此事的一些代码:

.channel(c -> c.queue())
.<Object>handle((p, h) -> databaseActivator.recordToDatabase(p, h), 
            e -> e.poller(p -> p.fixedDelay(0).maxMessagesPerPoll(1)))