使用 Reactive Kafka Binder 的基础示例
在本节中，我们将展示一些使用响应式绑定器编写响应式Kafka应用程序的基本代码片段及其相关细节。
@Bean
public Function<Flux<String>, Flux<String>> uppercase() {
return s -> s.map(String::toUpperCase);
}
你可以将上述 uppercase 函数与基于消息通道的 Kafka 绑定器（spring-cloud-stream-binder-kafka）以及本节讨论的反应式 Kafka 绑定器（spring-cloud-stream-binder-kafka-reactive）一起使用。当将此函数与常规 Kafka 绑定器一起使用时，尽管你在应用程序中使用了反应式类型（即在 uppercase 函数中），但你只能在函数执行期间获得反应式流。在函数执行上下文之外，由于底层绑定器并非基于反应式技术栈，因此无法获得反应式的好处。因此，尽管这看起来像是带来了完整的端到端反应式技术栈，但此应用程序仅是部分反应式的。
现在假设您正在使用适用于 Kafka 的响应式绑定器——spring-cloud-stream-binder-kafka-reactive 与上述函数应用配合使用。该绑定器实现将提供完整的响应式优势，从链路的顶端消费到底端发布全程如此。这是因为底层绑定器构建于 Reactor Kafka 的核心 API 之上。在消费者端，它利用了 KafkaReceiver——这是 Kafka 消费者的响应式实现。同样，在生产者端，它使用 KafkaSender API，即 Kafka 生产者的响应式实现。由于响应式 Kafka 绑定器的基础建立在正确的响应式 Kafka API 之上，应用程序能够充分利用响应式技术的全部优势。使用此响应式 Kafka 绑定器时，诸如自动背压等响应式能力已内置在应用中。
从 4.0.2 版本开始，你可以通过提供一个或多个 ReceiverOptionsCustomizer 或 SenderOptionsCustomizer bean 来分别自定义 ReceiverOptions 和 SenderOptions。它们是 BiFunction，接收绑定名称和初始选项，并返回自定义后的选项。这些接口扩展了 Ordered，因此当存在多个自定义器时，它们将按所需的顺序应用。
默认情况下，绑定器不会提交偏移量。从版本 4.0.2 开始，KafkaHeaders.ACKNOWLEDGMENT 标头包含一个 ReceiverOffset 对象，允许您通过调用其 acknowledge() 或 commit() 方法来提交偏移量。
@Bean
public Consumer<Flux<Message<String>>> consume() {
return msg -> {
process(msg.getPayload());
msg.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, ReceiverOffset.class).acknowledge();
}
}
有关更多信息，请参考 reactor-kafka 文档和 javadocs。
此外，从版本4.0.3开始，可以将Kafka消费者属性reactiveAtmostOnce设置为true，绑定器将在处理每次轮询返回的记录之前自动提交偏移量。同样，从版本4.0.3开始，您可以将消费者属性reactiveAutoCommit设置为true，绑定器将在处理每次轮询返回的记录之后自动提交偏移量。在这些情况下，确认头信息不会出现。
4.0.2 版本也提供了 reactiveAutoCommit，但其实现有误，其行为与 reactiveAtMostOnce 类似。
以下是使用 reactiveAutoCommit 的示例。
@Bean
Consumer<Flux<Flux<ConsumerRecord<?, String>>>> input() {
return flux -> flux
.doOnNext(inner -> inner
.doOnNext(val -> {
log.info(val.value());
})
.subscribe())
.subscribe();
}
请注意，在使用自动提交时，reactor-kafka 会返回一个 Flux<Flux<ConsumerRecord<?, ?>>>。由于 Spring 无法访问内部 flux 的内容，应用程序必须处理原生的 ConsumerRecord；不会对内容应用消息转换或转换服务。这需要使用原生解码（通过在配置中指定适当类型的 Deserializer）来返回所需类型的记录键/值。