如何使用Spring集成转换器处理在 Kafka 中接收的消息

Question

我将下面的数据发布到kafka并通过Spring集成通道接收并转换为Log对象，我如何使用Spring集成转换器将下面的数据转换为Log对象？感谢这里的任何帮助

'日志（客户端键=字符串，有效负载=字符串）”

这是通道适配器代码

@Bean
public KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter() {
    KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter<>(
            kafkaListenerContainer());
    kafkaMessageDrivenChannelAdapter.setPayloadType(Log.class);
    kafkaMessageDrivenChannelAdapter.setOutputChannel(inputChannel());
    return kafkaMessageDrivenChannelAdapter;
}

当我尝试使用下面的方法在服务激活器中进行转换时

 ObjectMapper objectMapper = new ObjectMapper();
 Log msg = objectMapper.readValue(arg0.getPayload().toString() , Log.class);

它的失败

com.fasterxml.jackson.core.JsonParseException： 无法识别的令牌“日志”：期待（“真”、“假”或“空”）

Answer 1

首先，'Log（clientKey=string， valload=string）'看起来不像您希望在Spring集成中从后面转换的格式良好的JSON。

另一个问题是，您的dokafkaMessageDrivenChannelAdapter.setPayloadType（Log.class）；。这样您就不需要任何下游转换，集成记录消息监听器中的消息转换器将为我们完成转换工作。

然而，我们仍然需要有一个适当的数据能够转换。

您还需要记住，Apache Kafka 有自己的机制来从线路中反序列化字节[]。有关转换和（反）序列化的更多信息，请参阅 Spring for Apache Kafka： https://docs.spring.io/spring-kafka/docs/current/reference/html/_reference.html#serdes

对于正确的JSON转换，Spring Integration提供了一个JsonToObjectTransformer组件，该组件可以与@Transformer注释一起使用：https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-transformation-chapter.html#_common_transformers

更新

@Transformer(inputChannel = "inputChannel", outputChannel = "processChannel")
@Bean
public JsonToObjectTransformer jsonToObjectTransformer() {
   return new JsonToObjectTransformer(Log.class);
}