spring整合kafka如何配置

在Spring Boot项目中整合Kafka，你需要遵循以下步骤进行配置：

1. 添加依赖

在你的pom.xml文件中添加Kafka和Spring Kafka的依赖：

<dependencies> <!-- Spring Kafka --> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <!-- Kafka Client --> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> </dependency> </dependencies> 

2. 配置Kafka

在application.properties或application.yml文件中配置Kafka相关信息，例如：

# Kafka服务器地址 spring.kafka.bootstrap-servers=localhost:9092  # Kafka主题 spring.kafka.consumer.group-id=my-group spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer  # Kafka生产者配置 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer 

3. 创建Kafka配置类

创建一个配置类，用于设置Kafka相关的Bean，例如：

@Configuration public class KafkaConfig { @Value("${spring.kafka.bootstrap-servers}") private String bootstrapServers; @Bean public Map<String, Object> consumerConfigs() { Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); return props; } @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, StringSerializer.class); return props; } @Bean public ConsumerFactory<String, String> consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } @Bean public ProducerFactory<String, String> producerFactory() { return new DefaultKafkaProducerFactory<>(producerConfigs()); } @Bean public KafkaTemplate<String, String> kafkaTemplate() { return new KafkaTemplate<>(producerFactory()); } } 

4. 创建Kafka监听器

创建一个Kafka监听器，用于处理接收到的消息，例如：

@Service public class KafkaListener { @KafkaListener(topics = "${spring.kafka.consumer.group-id}", groupId = "${spring.kafka.consumer.group-id}") public void listen(ConsumerRecord<String, String> record) { System.out.printf("Received message: key = %s, value = %s, partition = %d, offset = %d%n", record.key(), record.value(), record.partition(), record.offset()); } } 

5. 发送消息

创建一个Kafka生产者，用于发送消息到Kafka主题，例如：

@Service public class KafkaProducer { @Autowired private KafkaTemplate<String, String> kafkaTemplate; public void sendMessage(String topic, String message) { kafkaTemplate.send(topic, message); } } 

现在你已经成功地在Spring Boot项目中整合了Kafka，可以使用KafkaProducer发送消息，并使用KafkaListener监听和处理接收到的消息。