Spring 启动 2x + Kafka API 生成 JSON 对象而不是 String/Generic

Question

代码目前的样子：（模板）

@Configuration
public class KafkaConfiguration {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return props;
    }

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    public void send(String topicName, Object body) {
        kafkaTemplate.send(topic, body);
    }
    //In send method i am sending simple json in string format: {"name":"Leo", "age": "51"}
    //Would be better if i change object to JSON type and which type should   
    // i provide?
}

我应该以哪种方式发送此字符串？我应该使用像 JSONObject 这样的东西吗？或者我应该使用 ObjectMapper 从这个字符串中获取 json？

Answer 1

使用 JSON 有效负载生成消息始终是更好的方法。

您可以编写自定义方法将 String 序列化为 JSON

private final static ObjectWriter writer = new ObjectMapper().writerFor(TopicRecord.class);
private String serialize(TopicRecord topicRecord) {
    try {
      return writer.writeValueAsString(topicRecord);
    } catch (JsonProcessingException e) {
      throw new RuntimeException("Unable to serialize topic record to json", e);
    }
  }

Spring 启动 2x + Kafka API 生成 JSON 对象而不是 String/Generic

Spring boot 2x + Kafka API produce JSON object instead of String/Generic

java

spring-boot

apache-kafka

spring-kafka

kafka-producer-api