Spring集成：如何使用QueueChannel on-demand作为背压感知反应（流量）管道

Question

我有一个Spring集成队列通道，我想在背压感知流量管道中使用它。

1. 预取n来自队列的消息。
2. 使异步调用到外部系统像乐趣远程呼叫（消息：消息）：单声道

我不想使用带有调度程序的轮询器提前从队列中提取消息。

在最新的Spring集成Java/Kotlin DSL中，最好的方法是什么，包括错误恢复等？




Answer 1

请参见IntegrationReactiveUtils。messageChannelToFlux（）：

/**
 * Adapt a provided {@link MessageChannel} into a {@link Flux} source:
 * - a {@link org.springframework.integration.channel.FluxMessageChannel}
 * is returned as is because it is already a {@link Publisher};
 * - a {@link SubscribableChannel} is subscribed with a {@link MessageHandler}
 * for the {@link Sinks.Many#tryEmitNext(Object)} which is returned from this method;
 * - a {@link PollableChannel} is wrapped into a {@link MessageSource} lambda and reuses
 * {@link #messageSourceToFlux(MessageSource)}.
 * @param messageChannel the {@link MessageChannel} to adapt.
 * @param <T> the expected payload type.
 * @return a {@link Flux} which uses a provided {@link MessageChannel} as a source for events to publish.
 */
@SuppressWarnings("unchecked")
public static <T> Flux<Message<T>> messageChannelToFlux(MessageChannel messageChannel) {

然后您可以使用IntegrationFlows.from（Publisher）：

/**
 * Populate a {@link FluxMessageChannel} to the {@link IntegrationFlowBuilder} chain
 * and subscribe it to the provided {@link Publisher}.
 * @param publisher the {@link Publisher} to subscribe to.
 * @return new {@link IntegrationFlowBuilder}.
 */
public static IntegrationFlowBuilder from(Publisher<? extends Message<?>> publisher) {