java集成kafka案例

要在 Java 项目中集成 Apache Kafka 以实现消息的生产和消耗，步调如下：
1. 引入 Maven 依靠
在您的 pom.xml 文件中添加以下依靠，以包罗 Kafka 客户端库：

1. <dependencies>
2.     <!-- Kafka Clients -->
3.     <dependency>
4.         <groupId>org.apache.kafka</groupId>
5.         <artifactId>kafka-clients</artifactId>
6.         <version>2.8.0</version>
7.     </dependency>
8.     <!-- 如果使用 Spring Boot，可添加以下依赖 -->
9.     <dependency>
10.         <groupId>org.springframework.kafka</groupId>
11.         <artifactId>spring-kafka</artifactId>
12.         <version>2.7.0</version>
13.     </dependency>
14. </dependencies>

复制代码

2. 配置 Kafka 生产者
首先，设置生产者的配置属性：

1. import org.apache.kafka.clients.producer.KafkaProducer;
2. import org.apache.kafka.clients.producer.Producer;
3. import org.apache.kafka.clients.producer.ProducerConfig;
4. import org.apache.kafka.clients.producer.ProducerRecord;
5. import org.apache.kafka.common.serialization.StringSerializer;
7. import java.util.Properties;
9. public class KafkaProducerExample {
10.     public static void main(String[] args) {
11.         // 配置属性
12.         Properties props = new Properties();
13.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
14.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
15.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
17.         // 创建生产者
18.         Producer<String, String> producer = new KafkaProducer<>(props);
20.         // 发送消息
21.         for (int i = 0; i < 10; i++) {
22.             ProducerRecord<String, String> record = new ProducerRecord<>("your_topic", "key" + i, "value" + i);
23.             producer.send(record);
24.         }
26.         // 关闭生产者
27.         producer.close();
28.     }
29. }

复制代码

3. 配置 Kafka 消耗者
接下来，设置消耗者的配置属性，并订阅主题以消耗消息：

1. import org.apache.kafka.clients.consumer.Consumer;
2. import org.apache.kafka.clients.consumer.ConsumerConfig;
3. import org.apache.kafka.clients.consumer.ConsumerRecords;
4. import org.apache.kafka.clients.consumer.KafkaConsumer;
5. import org.apache.kafka.common.serialization.StringDeserializer;
7. import java.time.Duration;
8. import java.util.Collections;
9. import java.util.Properties;
11. public class KafkaConsumerExample {
12.     public static void main(String[] args) {
13.         // 配置属性
14.         Properties props = new Properties();
15.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
16.         props.put(ConsumerConfig.GROUP_ID_CONFIG, "your_group_id");
17.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
18.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
20.         // 创建消费者
21.         Consumer<String, String> consumer = new KafkaConsumer<>(props);
23.         // 订阅主题
24.         consumer.subscribe(Collections.singletonList("your_topic"));
26.         // 持续消费消息
27.         try {
28.             while (true) {
29.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
30.                 records.forEach(record -> {
31.                     System.out.printf("Consumed message: key = %s, value = %s, offset = %d%n",
32.                             record.key(), record.value(), record.offset());
33.                 });
34.             }
35.         } finally {
36.             // 关闭消费者
37.             consumer.close();
38.         }
39.     }
40. }

复制代码

4. 使用 Spring Boot 集成 Kafka
如果您使用 Spring Boot，可以通过配置 KafkaTemplate（用于生产消息）和使用 @KafkaListener 注解（用于消耗消息）来简化 Kafka 的集成。
生产者配置：

1. import org.apache.kafka.clients.producer.ProducerConfig;
2. import org.apache.kafka.common.serialization.StringSerializer;
3. import org.springframework.context.annotation.Bean;
4. import org.springframework.context.annotation.Configuration;
5. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
6. import org.springframework.kafka.core.KafkaTemplate;
7. import org.springframework.kafka.core.ProducerFactory;
9. import java.util.HashMap;
10. import java.util.Map;
12. @Configuration
13. public class KafkaProducerConfig {
15.     @Bean
16.     public ProducerFactory<String, String> producerFactory() {
17.         Map<String, Object> configProps = new HashMap<>();
18.         configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
19.         configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
20.         configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
21.         return new DefaultKafkaProducerFactory<>(configProps);
22.     }
24.     @Bean
25.     public KafkaTemplate<String, String> kafkaTemplate() {
26.         return new KafkaTemplate<>(producerFactory());
27.     }
28. }

复制代码

使用 KafkaTemplate 发送消息：

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.kafka.core.KafkaTemplate;
3. import org.springframework.stereotype.Service;
5. @Service
6. public class KafkaProducerService {
8.     @Autowired
9.     private KafkaTemplate<String, String> kafkaTemplate;
11.     public void sendMessage(String topic, String key, String value) {
12.         kafkaTemplate.send(topic, key, value);
13.     }
14. }

复制代码

消耗者配置：

1. import org.apache.kafka.clients.consumer.ConsumerConfig;
2. import org.apache.kafka.common.serialization.StringDeserializer;
3. import org.springframework.context.annotation.Bean;
4. import org.springframework.context.annotation.Configuration;
5. import org.springframework.kafka.annotation.EnableKafka;
6. import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
7. import org.springframework.kafka.core.ConsumerFactory;
8. import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
10. import java.util.HashMap;
11. import java.util.Map;
13. @EnableKafka
14. @Configuration
15. public class KafkaConsumerConfig {
17.     @Bean
18.     public ConsumerFactory<String, String> consumerFactory() {
19.         Map<String, Object> props = new HashMap<>();
20.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
21.         props.put(ConsumerConfig.GROUP_ID_CONFIG, "your_group_id");
22.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
23.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
24.         return new DefaultKafkaConsumerFactory<>(props);
25.     }
27.     @Bean
28.     public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
29.         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
30.         factory.setConsumerFactory(consumerFactory());
31.         return factory;
32.     }
33. }

复制代码

使用 @KafkaListener 消耗消息：
在 Spring Boot 中，@KafkaListener 注解用于监听指定的 Kafka 主题，并在收到消息时触发相应的方法。以下是一个基本示例：

1. import org.springframework.kafka.annotation.KafkaListener;
2. import org.springframework.stereotype.Service;
4. @Service
5. public class KafkaConsumerService {
7.     @KafkaListener(topics = "your_topic", groupId = "your_group_id")
8.     public void listen(String message) {
9.         System.out.println("Received message: " + message);
10.         // 在此处添加处理逻辑
11.     }
12. }

复制代码

在上述代码中：

• topics：指定要监听的 Kafka 主题。
  
• groupId：指定消耗者组 ID。
  

listen 方法：当有新消息发布到指定主题时，该方法会被调用，message 参数包罗消息的内容。
批量消耗消息
如果希望一次处理多条消息，可以启用批量监听。首先，必要配置一个支持批量消耗的 KafkaListenerContainerFactory：

1. import org.springframework.context.annotation.Bean;
2. import org.springframework.context.annotation.Configuration;
3. import org.springframework.kafka.annotation.EnableKafka;
4. import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
5. import org.springframework.kafka.core.ConsumerFactory;
7. @EnableKafka
8. @Configuration
9. public class KafkaConsumerConfig {
11.     @Bean
12.     public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
13.             ConsumerFactory<String, String> consumerFactory) {
14.         ConcurrentKafkaListenerContainerFactory<String, String> factory =
15.                 new ConcurrentKafkaListenerContainerFactory<>();
16.         factory.setConsumerFactory(consumerFactory);
17.         factory.setBatchListener(true); // 启用批量监听
18.         return factory;
19.     }
20. }

复制代码

然后，在消耗者服务中使用 @KafkaListener 注解，并指定使用上述配置的工厂：

1. import org.springframework.kafka.annotation.KafkaListener;
2. import org.springframework.stereotype.Service;
3. import java.util.List;
5. @Service
6. public class KafkaBatchConsumerService {
8.     @KafkaListener(
9.         topics = "your_topic",
10.         groupId = "your_group_id",
11.         containerFactory = "kafkaListenerContainerFactory"
12.     )
13.     public void listen(List<String> messages) {
14.         System.out.println("Received batch messages: " + messages);
15.         // 在此处添加批量处理逻辑
16.     }
17. }

复制代码

在上述代码中：

• containerFactory：指定使用支持批量消耗的工厂。
  

listen 方法的参数范例为 List<String>，用于接收一批消息。
控制消耗者的启动和停止
在某些情况下，大概必要在运行时控制 Kafka 消耗者的启动和停止。可以通过 KafkaListenerEndpointRegistry 来实现：

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.kafka.listener.KafkaListenerEndpointRegistry;
3. import org.springframework.kafka.listener.MessageListenerContainer;
4. import org.springframework.scheduling.annotation.Scheduled;
5. import org.springframework.stereotype.Service;
7. @Service
8. public class KafkaListenerManager {
10.     @Autowired
11.     private KafkaListenerEndpointRegistry registry;
13.     // 启动监听器
14.     public void startListener(String listenerId) {
15.         MessageListenerContainer listenerContainer = registry.getListenerContainer(listenerId);
16.         if (listenerContainer != null && !listenerContainer.isRunning()) {
17.             listenerContainer.start();
18.         }
19.     }
21.     // 停止监听器
22.     public void stopListener(String listenerId) {
23.         MessageListenerContainer listenerContainer = registry.getListenerContainer(listenerId);
24.         if (listenerContainer != null && listenerContainer.isRunning()) {
25.             listenerContainer.stop();
26.         }
27.     }
28. }

复制代码

在上述代码中：

• startListener 方法用于启动指定的监听器。
  
• stopListener 方法用于停止指定的监听器。
  
• listenerId 对应于 @KafkaListener 注解中的 id 属性。
  

通过这种方式，可以在应用运行时根据必要动态地控制 Kafka 消耗者的举动。
参考资料

• Spring for Apache Kafka: @KafkaListener 的使用示例（消耗）
  
• SpringBoot + Kafka 使用@KafkaListener注解批量消耗
  
• 在Spring Boot 中动态管理Kafka Listener
  

通过上述配置和代码示例，可以在 Spring Boot 项目中有用地集成 Kafka，实现消息的生产和消耗功能。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。