KafkaFactoryPool

Posted 2021-09-06 tonggc1668

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerConfig;
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 java.util.HashMap;
import java.util.Map;

@Configuration
@Data
@Slf4j
public class KafkaFactoryPool {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServer;
    @Value("${spring.kafka.producer.key-serializer}")
    private String producerKs;
    @Value("${spring.kafka.producer.value-serializer}")
    private String producerVs;

    @Bean
    public Map<String,Object> producerConfig(){
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServer);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, producerKs);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, producerVs);
        return props;
    }
    @Bean
    public ProducerFactory<String,Object> producerFactory(){
        DefaultKafkaProducerFactory<String, Object> producerFactory = new DefaultKafkaProducerFactory<>(producerConfig());
        producerFactory.setProducerPerThread(true);
        return producerFactory;
    }
    @Bean
    public KafkaTemplate<String,Object> kafkaTemplate(){
        log.info(bootstrapServer+"---"+producerKs+producerVs);
        return new KafkaTemplate<String, Object>(producerFactory());
    }
}

        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.3.6.RELEASE</version>
        </dependency>