如何使用 Spring Cloud Stream 处理反序列化异常并转换为新模式？答案

【问题标题】：How to Handle Deserialization Exception & Converting to New Schema with Spring Cloud Stream?如何使用 Spring Cloud Stream 处理反序列化异常并转换为新模式？
【发布时间】：2021-03-10 20:09:25
【问题描述】：

我无法理解如何在 Spring Cloud 流中正确处理反序列化异常。主要是因为实现的框架不支持标头，并且 DLQ 应该是与原始消息不同的模式。所以处理流程需要是：消费消息 -> 反序列化错误 -> DlqHandler -> 使用 NEW 模式序列化 -> 发送到 DLQ

下面链接的文档并不能很好地说明这是否可能。我已经看过很多 Spring-Kafka 的 SeekToCurrentErrorHandler 示例，但据我所知，这些示例是不同的实现，与我如何正确获取反序列化错误然后有一个自定义代码部分序列化为新格式并移动不匹配从那里开始。

我的主要问题是：是否可以使用 Spring Cloud Stream (kafka) 捕获反序列化异常并重新序列化？

Spring Cloud Documentation for DLQ

【问题讨论】：

标签： spring-cloud spring-cloud-stream dead-letter spring-cloud-stream-binder-kafka

【解决方案1】：

是，但不使用绑定重试或 DLQ 属性。

相反，添加一个ListenerContainerCustomizer bean 并使用为您需要的重试配置的SeekToCurrentErrorHandler 自定义绑定的侦听器容器，并且可能是DeadLetterPublishingRecoverer 的子类，使用适当配置的KafkaTemplate 并可能覆盖createProducerRecord 方法。

【讨论】：