spring boot 配置使用kafka

 

 

启动kafka正常

 

1.创建一个spring boot项目

 

2.配置使用kafka

引入依赖（gradle）

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

 

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

KafkaConfig.java

@Configuration
@EnableKafka
public 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);
    }
}

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

发送消息可以

@Autowired
private KafkaTemplate kafkaTemplate;

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

这样去测试使用

 

一个基础spring boot 配置kafka