我们很高兴地宣布发布 Reactor Kafka 1.0.0 的第一个里程碑版本。

什么是 Reactor Kafka？

Reactor Kafka 是一个基于 Project Reactor 的 Apache Kafka 响应式 API。Reactor Kafka API 能够使用具有非阻塞背压和极低开销的函数式 API 将消息发布到 Kafka 主题并从 Kafka 主题消费消息。这使得使用 Reactor 的应用程序能够将 Kafka 用作消息总线或流平台，并与其他系统集成以提供端到端的响应式管道。

Reactor Kafka 的价值主张是在具有多个外部交互（其中 Kafka 是外部系统之一）的应用程序中高效利用资源。端到端响应式管道受益于非阻塞背压和高效的线程使用，从而能够高效处理大量并发请求。Project Reactor 提供的优化使响应式应用程序的开发具有极低的开销和可预测的容量规划，从而提供低延迟、高吞吐量的管道。

入门

要开始运行响应式 Kafka 生产者和消费者示例，请遵循参考指南中“入门”部分的说明。

Reactor Kafka API

Reactor Kafka API 基于 Apache Kafka 生产者/消费者 API，由两个主要类组成

• Sender 用于向 Kafka 主题发布消息

• Receiver 用于从 Kafka 主题消费消息

这些响应式接口提供了底层 Kafka Producer 和 Consumer 的完整功能。

响应式发送器

Sender<Integer, String> sender =
    Sender.create(SenderOptions.create(producerProps));                 (1)
Flux<SenderRecord<Integer, String, Integer>> outboundFlux =             (2)
    Flux.range(1, 10)
        .map(i -> SenderRecord.create(producerRecord(topic, i), i));
sender.send(outboundFlux, false)                                        (3)
      .doOnNext(r -> log.debug("Message #{} result: {}",
                         r.correlationMetadata(), r.recordMetadata()))  (4)
      .subscribe();                                                     (5)

1. 创建 Sender

2. 要发送到 Kafka 的出站消息的 Flux

3. 响应式发送

4. 记录每次发送的结果

5. 订阅以启动消息流到 Kafka

响应式接收器

ReceiverOptions<Integer, String> receiverOptions =                      (1)
    ReceiverOptions.<Integer, String>create(consumerProps)
                   .subscription(Collections.singleton(topic));
Receiver.create(receiverOptions)                                        (2)
        .receive()                                                      (3)
        .subscribe(r -> {
                log.info("Received message {} ", r.record());           (4)
                r.offset().acknowledge();                               (5)
            });

1. 创建 ReceiverOptions 并配置对 Kafka 主题的订阅

2. 创建 Receiver

3. 响应式接收

4. 记录每条传入消息

5. 处理消息后确认，以便可以提交偏移量

资源

Reactor Kafka 源码和示例可在 github 上获取。

有关更多信息和其他资源，请参阅 Reactor Kafka 参考指南和 Javadocs。