spring boot 配置使用kafka

启动kafka正常

1.创建一个spring boot项目

2.配置使用kafka

引入依赖（gradle）

implementation ‘org.springframework.kafka:spring-kafka‘

添加配置文件（全部使用java配置）

KafkaConfig.java

@Configuration@EnableKafkapublic class KafkaConfig {    private static final String KAFKA_SERVERS_CONFIG = "127.0.0.1:9092";    private static final String LOCAL_GROUP_ID = "test";

    @Bean    ConcurrentKafkaListenerContainerFactory<Integer, String>    kafkaListenerContainerFactory() {        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =                new ConcurrentKafkaListenerContainerFactory<>();        factory.setConsumerFactory(consumerFactory());        return factory;    }

    @Bean    public ConsumerFactory<Integer, String> consumerFactory() {        return new DefaultKafkaConsumerFactory<>(consumerConfigs());    }

    @Bean    public Map<String, Object> consumerConfigs() {        Map<String, Object> props = new HashMap<>();        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS_CONFIG);        props.put(ConsumerConfig.GROUP_ID_CONFIG, LOCAL_GROUP_ID);        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);        return props;    }

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

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

    @KafkaListener(topics = "TEST_TOPIC_NEW")    public void listen(String data) {        System.out.println("kafkaconfig =listen======="+data);    }}

配置类中配置了消费者和生产者

发送消息可以

@Autowiredprivate KafkaTemplate kafkaTemplate;

@RequestMapping("/admin/kafka/send")public String sendMsg(String msg){    kafkaTemplate.send("TEST_TOPIC_NEW", msg);    return "success";}

这样去测试使用

一个基础spring boot 配置kafka

原文地址：https://www.cnblogs.com/guochaolang/p/12031116.html