数据仓库与分析java集成kafka案例 - Powered by Discuz! Archiver

万有斥力 发表于 2025-4-9 09:46:01

java集成kafka案例

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

<dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>2.8.0</version>
</dependency>

<dependency>
   <groupId>org.springframework.kafka</groupId>
   <artifactId>spring-kafka</artifactId>
   <version>2.7.0</version>
</dependency>
</dependencies> 2. 配置 Kafka 生产者
首先，设置生产者的配置属性：
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerExample {
public static void main(String[] args) {
   // 配置属性
   Properties props = new Properties();
   props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
   props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

   // 创建生产者
   Producer<String, String> producer = new KafkaProducer<>(props);

   // 发送消息
   for (int i = 0; i < 10; i++) {
         ProducerRecord<String, String> record = new ProducerRecord<>("your_topic", "key" + i, "value" + i);
         producer.send(record);
   }

   // 关闭生产者
   producer.close();
}
} 3. 配置 Kafka 消耗者
接下来，设置消耗者的配置属性，并订阅主题以消耗消息：
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
public static void main(String[] args) {
   // 配置属性
   Properties props = new Properties();
   props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   props.put(ConsumerConfig.GROUP_ID_CONFIG, "your_group_id");
   props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
   props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

   // 创建消费者
   Consumer<String, String> consumer = new KafkaConsumer<>(props);

   // 订阅主题
   consumer.subscribe(Collections.singletonList("your_topic"));

   // 持续消费消息
   try {
         while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            records.forEach(record -> {
               System.out.printf("Consumed message: key = %s, value = %s, offset = %d%n",
                        record.key(), record.value(), record.offset());
            });
         }
   } finally {
         // 关闭消费者
         consumer.close();
   }
}
} 4. 使用 Spring Boot 集成 Kafka
如果您使用 Spring Boot，可以通过配置 KafkaTemplate（用于生产消息）和使用 @KafkaListener 注解（用于消耗消息）来简化 Kafka 的集成。
生产者配置：
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaProducerConfig {

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

@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
   return new KafkaTemplate<>(producerFactory());
}
} 使用 KafkaTemplate 发送消息：
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void sendMessage(String topic, String key, String value) {
   kafkaTemplate.send(topic, key, value);
}
} 消耗者配置：
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

@Bean
public ConsumerFactory<String, String> consumerFactory() {
   Map<String, Object> props = new HashMap<>();
   props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   props.put(ConsumerConfig.GROUP_ID_CONFIG, "your_group_id");
   props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
   props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
   return new DefaultKafkaConsumerFactory<>(props);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
   ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
   factory.setConsumerFactory(consumerFactory());
   return factory;
}
}
使用 @KafkaListener 消耗消息：
在 Spring Boot 中，@KafkaListener 注解用于监听指定的 Kafka 主题，并在收到消息时触发相应的方法。以下是一个基本示例：
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

@KafkaListener(topics = "your_topic", groupId = "your_group_id")
public void listen(String message) {
   System.out.println("Received message: " + message);
   // 在此处添加处理逻辑
}
}

在上述代码中：

[*]topics：指定要监听的 Kafka 主题。
[*]groupId：指定消耗者组 ID。
listen 方法：当有新消息发布到指定主题时，该方法会被调用，message 参数包罗消息的内容。
批量消耗消息
如果希望一次处理多条消息，可以启用批量监听。首先，必要配置一个支持批量消耗的 KafkaListenerContainerFactory：
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
         ConsumerFactory<String, String> consumerFactory) {
   ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
   factory.setConsumerFactory(consumerFactory);
   factory.setBatchListener(true); // 启用批量监听
   return factory;
}
} 然后，在消耗者服务中使用 @KafkaListener 注解，并指定使用上述配置的工厂：
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import java.util.List;

@Service
public class KafkaBatchConsumerService {

@KafkaListener(
   topics = "your_topic",
   groupId = "your_group_id",
   containerFactory = "kafkaListenerContainerFactory"
)
public void listen(List<String> messages) {
   System.out.println("Received batch messages: " + messages);
   // 在此处添加批量处理逻辑
}
} 在上述代码中：

[*]containerFactory：指定使用支持批量消耗的工厂。
listen 方法的参数范例为 List<String>，用于接收一批消息。
控制消耗者的启动和停止
在某些情况下，大概必要在运行时控制 Kafka 消耗者的启动和停止。可以通过 KafkaListenerEndpointRegistry 来实现：
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.listener.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

@Service
public class KafkaListenerManager {

@Autowired
private KafkaListenerEndpointRegistry registry;

// 启动监听器
public void startListener(String listenerId) {
   MessageListenerContainer listenerContainer = registry.getListenerContainer(listenerId);
   if (listenerContainer != null && !listenerContainer.isRunning()) {
         listenerContainer.start();
   }
}

// 停止监听器
public void stopListener(String listenerId) {
   MessageListenerContainer listenerContainer = registry.getListenerContainer(listenerId);
   if (listenerContainer != null && listenerContainer.isRunning()) {
         listenerContainer.stop();
   }
}
} 在上述代码中：

[*]startListener 方法用于启动指定的监听器。
[*]stopListener 方法用于停止指定的监听器。
[*]listenerId 对应于 @KafkaListener 注解中的 id 属性。
通过这种方式，可以在应用运行时根据必要动态地控制 Kafka 消耗者的举动。
参考资料

[*]Spring for Apache Kafka: @KafkaListener 的使用示例（消耗）
[*]SpringBoot + Kafka 使用@KafkaListener注解批量消耗
[*]在Spring Boot 中动态管理Kafka Listener
通过上述配置和代码示例，可以在 Spring Boot 项目中有用地集成 Kafka，实现消息的生产和消耗功能。

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

页: [1]

IT评测·应用市场-qidao123.com技术社区's Archiver

java集成kafka案例