java - spring cloud stream kafka背压

Question

如何在 spering 云流中使用带有 kafka binder 的项目 react 堆背压功能？

@Bean
public Function<Flux<String>, Flux<String>> processor() {
    return flux -> flux.delayElements(Duration.ofSeconds(1));
}

如果我使用这种方式，发布者会延迟 1 秒发送消息，但消费者会毫无延迟地消费消息。

可以在 cunsumer 上使用 spring cloud stream 使用 backperessure 吗？

Answer 1

你不能。背压意味着消息的生产者(从代理中出队)知道背压，在这种情况下它不是。它依赖于来自 Spring Integration 框架的入站 channel 适配器。为了支持背压，这些消息代理需要真正的响应式(Reactive) API/方法，例如 some initiatives in RabbitMQ。