SpringBoot集成Kafka 配置工具类的详细代码

下面是详细讲解SpringBoot集成Kafka配置工具类的完整攻略。

1、前置要求

在进行SpringBoot集成Kafka之前，需要准备以下环境：

• Java JDK 8及以上版本
• Maven构建工具
• Kafka集群及对应的Zookeeper集群

2、添加依赖

在进行SpringBoot集成Kafka之前，需要在pom.xml中添加以下依赖：

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

3、配置Kafka连接信息

在SpringBoot项目中，需要在application.yml或application.properties中配置Kafka的连接信息，例如：

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=group1

4、编写Kafka配置工具类

Kafka配置工具类的作用是通过@Configuration注解将Kafka配置信息注入到Spring容器中，该工具类中需要包含以下内容：

• 配置Kafka Producer
• 配置Kafka Consumer

下面是一个完整的Kafka配置工具类代码示例：

@Configuration
public class KafkaConfig {
    @Autowired
    private KafkaProperties kafkaProperties;

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

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

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setErrorHandler(new SeekToCurrentErrorHandler());
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getConsumer().getGroupId());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaProperties.getConsumer().getAutoOffsetReset());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }
}

5、使用Kafka配置工具类

在进行Kafka开发时，需要在对应的生产者或者消费者中使用Kafka配置工具类来进行Kafka生产和消费。例如，下面是一个基于Kafka的生产者代码示例：

@Service
public class KafkaProducerService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

下面是一个基于Kafka的消费者代码示例：

@Service
public class KafkaConsumerService {
    @KafkaListener(topics = "test_topic")
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

在上述代码示例中，生产者使用了@Autowired注解进行KafkaTemplate的注入，并通过KafkaTemplate.send()方法来发送消息。消费者使用了@KafkaListener注解来监听指定的Kafka主题，并在监听到消息时进行业务处理。

至此，SpringBoot集成Kafka配置工具类的完整攻略就讲解完了，希望对你有所帮助。