如何在Spring Reactor-Kafka中处理反序列化错误？

现有方案的问题 #

你当前手动在flatMap中完成反序列化的方式虽然能处理异常，但存在耦合问题——将反序列化逻辑与业务消费逻辑绑定，没有利用Kafka原生的反序列化机制，扩展性和可维护性较差。下面提供两种更优雅的解决方案，基于Spring Kafka的JsonDeserializer或自定义反序列化器来统一处理反序列化异常。

方案一：使用Spring Kafka JsonDeserializer + Reactor错误处理 #

1. 配置JsonDeserializer #

先在消费者属性中配置JsonDeserializer，并开启反序列化失败时抛出异常的配置：

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Toy.class); // 指定默认反序列化类型
props.put(JsonDeserializer.TRUSTED_PACKAGES, "你的Toy类所在包路径"); // 配置信任包
props.put(JsonDeserializer.FAIL_ON_UNKNOWN_PROPERTIES, true); // 遇到未知属性时抛出异常

2. 消费流中捕获并处理异常 #

在Reactor消费流中，利用onErrorContinue或onErrorResume捕获反序列化异常，提取原始消息后调用REST接口处理：

reactiveKafkaConsumerTemplate
        .receiveAutoAck() // 自动提交offset，若需控制重试可改用receive()手动ack
        .onErrorContinue((throwable, record) -> {
            // 判断是否为反序列化异常
            if (throwable instanceof SerializationException) {
                // 从原始字节提取消息内容（反序列化失败时record.value()可能为null）
                String rawMessage;
                try {
                    rawMessage = new String(record.valueBytes(), StandardCharsets.UTF_8);
                } catch (Exception e) {
                    rawMessage = "无法解析原始消息字节";
                }
                // 调用REST接口处理失败消息，非阻塞式订阅
                handleError(rawMessage).subscribe();
            }
            // 其他类型异常可按需扩展处理逻辑
        })
        .flatMap(record -> handle(record.value())) // 反序列化成功后执行业务逻辑
        .subscribe();

如果需要更精细的offset控制（比如失败时不提交offset以便重试），可以改用receive()手动管理ack：

reactiveKafkaConsumerTemplate
        .receive()
        .flatMap(record -> {
            return Mono.just(record.value())
                    .flatMap(toy -> handle(toy)
                            .doOnSuccess(v -> record.receiverOffset().acknowledge())) // 成功则提交offset
                    .onErrorResume(e -> {
                        if (e instanceof SerializationException) {
                            String rawMessage = new String(record.valueBytes(), StandardCharsets.UTF_8);
                            return handleError(rawMessage)
                                    .doOnSuccess(v -> record.receiverOffset().acknowledge()) // 处理完错误后提交offset
                                    .doOnError(err -> record.receiverOffset().nack()); // 若错误处理失败，拒绝消息触发重试
                        }
                        return Mono.error(e);
                    });
        })
        .subscribe();

方案二：自定义反序列化器 #

如果JsonDeserializer无法满足需求，可以自定义反序列化器，在其中抛出异常后由Reactor流统一处理：

1. 实现自定义反序列化器 #

public class ToyDeserializer implements Deserializer<Toy> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 可添加自定义配置，比如设置序列化特性
    }

    @Override
    public Toy deserialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }
        try {
            return objectMapper.readValue(data, Toy.class);
        } catch (IOException e) {
            // 抛出序列化异常，交由Reactor流处理
            throw new SerializationException("反序列化Toy失败: " + new String(data), e);
        }
    }

    @Override
    public void close() {
        // 资源清理操作
    }
}

2. 配置并使用自定义反序列化器 #

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ToyDeserializer.class);

后续的消费流异常处理逻辑与方案一完全一致，无需额外修改。

方案对比 #

• 现有手动反序列化：耦合业务与反序列化逻辑，重复代码多，不推荐。
• JsonDeserializer/自定义反序列化器：符合Kafka原生设计，解耦反序列化与业务逻辑，异常处理更集中，扩展性更强，是更优的选择。

内容的提问来源于stack exchange，提问作者Inessa