为什么需要衔接
eventd v2（卷二）是响应式 HTTP 服务，eventd v3（卷三）是流处理引擎。两者通过 Kafka 衔接：
eventd v2 (Reactor Netty + WebFlux + R2DBC)
↓ 业务事件
Reactive Kafka Producer (Mono → KafkaRecord)
↓
Kafka Topic "eventd-raw"
↓
Kafka Streams / Flink (eventd v3)
v2 用 Mono/Flux 处理请求，需要把结果异步写入 Kafka。
Reactive Kafka 生产者
使用 reactor-kafka（Project Reactor 生态的 Kafka 响应式客户端，与 Spring 无强绑定）：
<dependency>
<groupId>io.projectreactor.kafka</groupId>
<artifactId>reactor-kafka</artifactId>
<version>1.3.25</version>
</dependency>
import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderOptions;
import reactor.kafka.sender.SenderRecord;
public class EventPublisher {
private final KafkaSender<String, String> sender;
public EventPublisher(String bootstrapServers) {
SenderOptions<String, String> options = SenderOptions.<String, String>create()
.producerProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
.producerProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class)
.producerProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class)
.producerProperty(ProducerConfig.ACKS_CONFIG, "all");
this.sender = KafkaSender.create(options);
}
public Mono<Void> publish(String key, String eventJson) {
SenderRecord<String, String, String> record = SenderRecord.create(
new ProducerRecord<>("eventd-raw", key, eventJson),
key // 关联数据（用于确认回调）
);
return sender.send(Mono.just(record))
.next() // 只取第一个结果
.flatMap(result -> result.exception() == null
? Mono.just(result)
// 发送失败不会让 Flux 报错，只体现在 SenderResult 里；
// 必须显式转成 error，后续的 retryWhen / onErrorResume 才有机会执行
: Mono.error(new KafkaException("send failed", result.exception())))
.then();
}
}
这里最容易写错的一点：sender.send(...) 返回的 Flux<SenderResult> 不会因为单条发送失败而 onError。只打印 result.exception() 会让调用方以为发送成功，事务回滚或死信补偿都不会触发。必须像上面这样把异常转成 Mono.error。
WebFlux 中集成
@Service
public class OrderService {
private final OrderRepository orderRepo;
private final EventPublisher eventPublisher;
public Mono<Order> createOrder(OrderRequest request) {
return orderRepo.save(request.toOrder())
.flatMap(order -> {
String eventJson = toEventJson(order);
return eventPublisher.publish(order.id(), eventJson)
.thenReturn(order);
});
}
private String toEventJson(Order order) {
return "{\"eventId\":\"" + UUID.randomUUID() + "\"," +
"\"type\":\"ORDER\"," +
"\"userId\":\"" + order.userId() + "\"," +
"\"amount\":" + order.amount() + "," +
"\"timestamp\":" + System.currentTimeMillis() + "}";
}
}
publish 返回 Mono<Void>，可以无缝接入 WebFlux 的响应式链。
注意事务边界：如果外层用 TransactionalOperator 包住整条链，flatMap 里的 publish 会在事务提交前执行，与后文「提交后再发事件」的约定相反。要让提交先行，事务只能包住 save：
return txOperator.transactional(orderRepo.save(request.toOrder()))
.flatMap(order -> eventPublisher.publish(order.id(), toEventJson(order))
.thenReturn(order));
背压衔接
从 WebFlux 到 Kafka 的背压如何传递？
WebFlux 请求 → Mono → Reactive Kafka Producer → Kafka ProducerBuffer
↑ ↓
└──── request(n) ← 背压 ← send ack 速率 ───────┘
reactor-kafka 的 KafkaSender 有最大在途请求数上限（SenderOptions.maxInFlight()，不同版本默认值不同，1.3.x 是 1024），确认回来之前不会无限往里塞。容量按业务延迟要求显式设置比依赖默认值更可靠。上游把 publish() 返回的 Mono 串进请求链（flatMap），broker 变慢时这条链才会真的降速。
但这条背压链只在调用方等待结果时成立。如果写成 publish(...).subscribe() 然后立即返回响应，WebFlux 就不受任何约束——这正是响应式代码里最常见的「以为有背压，其实没有」。
Kafka Streams 读取 Reactive 生产的数据
Kafka Streams 不感知生产者是 Reactive 还是阻塞式——它只消费 Kafka Topic 中的消息。
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> rawEvents = builder.stream("eventd-raw");
KStream<String, String> cleanedEvents = rawEvents
.filter((key, value) -> isValidJson(value))
.mapValues(this::normalizeEvent);
cleanedEvents.to("eventd-cleaned");
错误处理
生产事件到 Kafka 可能失败（broker 不可用、认证失败）。处理策略：
public Mono<Order> createOrder(OrderRequest request) {
return orderRepo.save(request.toOrder())
.flatMap(order ->
eventPublisher.publish(order.id(), toEventJson(order))
.retryWhen(Retry.backoff(3, Duration.ofMillis(100))
.filter(err -> err instanceof KafkaException))
.onErrorResume(err -> {
// 写入本地死信队列（稍后重试）
deadLetterQueue.save(order.id(), toEventJson(order), err);
return Mono.empty();
})
.thenReturn(order)
);
}
事件格式约定
v2 写入 eventd-raw 的就是扁平结构，字段名用 timestamp：
{
"eventId": "uuid",
"type": "ORDER",
"userId": "alice",
"amount": 100.0,
"timestamp": 1700000000000
}
eventId：全局唯一，用于去重type：事件类型，大写（下游按ORDER/REFUND过滤）userId、amount：业务维度timestamp：v2 的事件时间（毫秒 Unix 时间戳）
v3 的清洗管道会把它转成 eventTime 并补上 version、source、symbol 等字段，转换逻辑与兼容写法见 14.3 从 v2 到 v3 的演进
。全卷后续章节（9.3、12.x、13.x）统一使用 eventTime。
与卷二衔接
eventd v2 的 OrderService.createOrder() 在 R2DBC 事务提交后，异步发送事件到 Kafka：
OrderService.createOrder()
→ R2DBC TransactionalOperator（事务提交）
→ EventPublisher.publish()（Reactive Kafka）
→ Kafka Topic "eventd-raw"
↓
eventd v3 Kafka Streams 消费
关键点：事件发送在数据库事务提交后进行。如果发送失败，订单已在数据库中，但事件未发出——由死信队列补偿。
小结
| 组件 | 角色 |
|---|---|
reactor-kafka | Reactive Kafka 生产者/消费者 |
KafkaSender.send() | Flux<SenderRecord> → Flux<SenderResult> |
| 背压 | 生产确认速率限制上游 |
| 错误处理 | 重试 + 死信队列 |
| 事件格式 | v2：JSON，含 eventId/type/userId/amount/timestamp；v3 追加 version/source/symbol 与 eventTime |
第 3 章结束，第一部分（Kafka Streams）完成。下一节 4.1 DataStream API：Source/Transformation/Sink 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。