使用 Reactor Kafka 时，可通过 mapNotNull() 运算符安全跳过格式不合法的 JSON 消息（如截断、非法字符等），避免抛异常、返回 null 或默认对象，实现真正的“无操作”处理。
使用 reactor kafka 时，可通过 `mapnotnull()` 运算符安全跳过格式不合法的 json 消息（如截断、非法字符等），避免抛异常、返回 null 或默认对象，实现真正的“无操作”处理。
在基于 Project Reactor 的 Kafka 消费链路中，Flux<t></t> 的每个阶段都应遵循响应式流契约：不能中断流（除非错误），也不能注入 null 值（会触发 NullPointerException）。你原始代码中用 map() + return null 的方式不仅违反规范，还会导致运行时崩溃；而直接抛出 RuntimeException 则会使整个消费者中断——这不符合“仅忽略损坏消息”的业务诉求。
正确做法是改用 mapNotNull() ——这是 Reactor 提供的专用运算符，语义明确：当映射函数返回 null 时，该元素将被静默过滤（dropped），不会进入后续流程，也不会终止流。它既不传播错误，也不污染数据流，完美契合“do nothing”场景。
以下是优化后的完整示例：
Flux<Person> consume() {
return kafkaReceiver.receive()
.mapNotNull(record -> {
try {
// ✅ 正常 JSON → Person（成功则返回实例）
return objectMapper.readValue(record.value(), Person.class);
} catch (JsonEOFException | JsonParseException e) {
// ⚠️ JSON 截断或语法错误（如 {"name":"John", "ag）→ 忽略该消息
LOGGER.warn("Skipping malformed JSON record (incomplete/invalid syntax): {}", record.value(), e);
return null; // ← mapNotNull 会自动丢弃此项
} catch (JsonMappingException e) {
// ❌ 字段类型不匹配、缺失必需字段等 → 视为业务错误，需中断并告警
LOGGER.error("Invalid JSON structure for Person: {}", record.value(), e);
throw new RuntimeException("Invalid Person payload", e);
} catch (IOException e) {
// ? 其他 I/O 异常（非 JSON 本身问题）→ 不建议忽略，应传播
throw Exceptions.propagate(e);
}
})
.doOnNext(person -> doSomething(person)); // 后续业务逻辑
}⚠️ 关键注意事项：
-
mapNotNull()是map()的安全替代，仅对null返回值做静默过滤，其他任何异常仍会向下游传播； -
JsonEOFException（来自 Jackson）和JsonParseException是识别“不完整 JSON”的典型异常，应归入忽略范畴；但JsonMappingException表示 JSON 合法但语义错误（如 age 字段为字符串），通常需显式处理； - 切勿在
mapNotNull内部捕获Throwable或吞掉所有异常——这会掩盖真实故障； - 若需监控丢弃率，可在
return null前增加指标埋点（如 Micrometer Counter）； - Kafka 位移提交策略需与之匹配：若使用自动提交，被
mapNotNull过滤的消息仍会被提交 offset；若需精确一次（exactly-once），建议启用手动提交并结合checkpoint()在doOnNext后确认。
总结：mapNotNull() 是 Reactor 生态中实现“条件性跳过”的标准、声明式、零副作用方案。它让错误处理更清晰、流更健壮，是构建高可用 Kafka 消费器的关键实践之一。
