Spring Boot Kafka通用JSON模板

如何解决Spring Boot Kafka通用JSON模板

我正在使用kafka和spring boot，我需要将JSON对象发送到kafka，这一点是，我可以将配置KafkaTemplate的对象作为JSON发送，但仅用于该对象。

package com.bankia.apimanager.config;

import com.bankia.apimanager.model.RequestDTO;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;

@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,RequestDTO> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

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

}

package com.bankia.apimanager.controller;

import com.bankia.apimanager.model.RequestDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/infrastructure")
public class InfraStructureRequestController {

    private final static Logger LOG = LoggerFactory.getLogger( InfraStructureRequestController.class );

    private static final String TOPIC = "test";

    @Autowired
    private KafkaTemplate<String,RequestDTO> sender;

    @RequestMapping(value = "/test",method = RequestMethod.GET)
    public String postMessage(){

        ListenableFuture<SendResult<String,RequestDTO>> future = sender.send(TOPIC,new RequestDTO("Hola","Paco"));
        future.addCallback(new ListenableFutureCallback<SendResult<String,RequestDTO>>() {
            @Override
            public void onSuccess(SendResult<String,RequestDTO> result) {
                LOG.info("Sent message with offset=[" + result.getRecordMetadata().offset() + "]");
            }
            @Override
            public void onFailure(Throwable ex) {
                LOG.error("Unable to send message due to : " + ex.getMessage());
            }
        });
        return "OK";
    }
}

但是如果现在我想发送一个新的DTO对象怎么办？我必须声明一个新的KafkaTemplate<String,NEWOBJECT>并自动装配在配置中为每个对象声明的每个kafka模板吗？还有一种方法可以只声明一个kafkaTemplate，在其中我可以发送任何类型的对象，并且将自动以JSON进行序列化？

解决方法

我认为，您可以指定通用的KafkaTemplate<String,Object>并将生产者值序列化器设置为JsonSerializer，如下所示：

@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());
    }

}

,

引用您的代码：

• 正确将Value Serializer定义为JsonSerializer，它将任何类型的对象转换为JSON。

@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;
}

• 在KafkaConfig和Controller中的每个位置，将更改为。

请记住，泛型会一直保留到编译时（类型擦除）