脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → java集成kafka

java集成kafka实例代码

 更新时间：2024年12月30日 14:15:59   作者：沉墨的夜  
文章介绍了如何在Java项目中集成Apache Kafka以实现消息的生产和消费,通过添加Maven依赖、配置生产者和消费者、使用SpringBoot简化集成以及控制消费者的启动和停止,可以实现高效的消息处理

java集成kafka

要在 Java 项目中集成 Apache Kafka 以实现消息的生产和消费，步骤如下：

1. 引入 Maven 依赖

在您的 pom.xml 文件中添加以下依赖，以包含 Kafka 客户端库：

<dependencies>
    <!-- Kafka Clients -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.8.0</version>
    </dependency>
    <!-- 如果使用 Spring Boot，可添加以下依赖 -->
    <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 Boot 项目中有效地集成 Kafka，实现消息的生产和消费功能。

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

相关文章

  • Java动态代理静态代理实例分析

    Java动态代理静态代理实例分析

    这篇文章主要介绍了Java动态代理静态代理实例分析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-03-03
  • Spring Boot与前端配合与Idea配置部署操作过程

    Spring Boot与前端配合与Idea配置部署操作过程

    这篇文章主要介绍了Spring Boot与前端配合与Idea配置部署的操作过程，本文图文并茂给大家介绍的非常详细,需要的朋友可以参考下
    2018-02-02
  • Java常用占位符方法简单代码实例

    Java常用占位符方法简单代码实例

    占位符是Java中常用的技术,用于在字符串中插入变量值或动态生成字符串,这篇文章主要给大家介绍了关于Java常用占位符方法的相关资料,文中介绍的非常详细,需要的朋友可以参考下
    2024-01-01
  • SpringMVC中Model与Session的区别说明

    SpringMVC中Model与Session的区别说明

    这篇文章主要介绍了SpringMVC中Model与Session的区别说明，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-12-12
  • 基于ScheduledExecutorService的两种方法(详解)

    基于ScheduledExecutorService的两种方法(详解)

    下面小编就为大家带来一篇基于ScheduledExecutorService的两种方法(详解)。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
    2017-10-10
  • 深入理解Java设计模式之中介者模式

    深入理解Java设计模式之中介者模式

    这篇文章主要介绍了JAVA设计模式之中介者模式的的相关资料，文中示例代码非常详细，供大家参考和学习，感兴趣的朋友可以了解
    2021-11-11
  • springboot整合RabbitMQ中死信队列的实现

    springboot整合RabbitMQ中死信队列的实现

    死信是无法被消费的消息,产生原因包括消息TTL过期、队列最大长度达到以及消息被拒绝且不重新排队,RabbitMQ的死信队列机制能够有效防止消息数据丢失,适用于订单业务等场景,本文就来介绍一下
    2024-10-10
  • SpringSecurity的@EnableWebSecurity注解详解

    SpringSecurity的@EnableWebSecurity注解详解

    这篇文章主要介绍了SpringSecurity的@EnableWebSecurity注解详解,@EnableWebSecurity是开启SpringSecurity的默认行为,它的上面有一个Import注解导入了WebSecurityConfiguration类,就是往IOC容器中注入了WebSecurityConfiguration这个类,需要的朋友可以参考下
    2023-11-11
  • IDEA EasyCode 一键帮你生成所需代码

    IDEA EasyCode 一键帮你生成所需代码

    这篇文章主要介绍了IDEA EasyCode 一键帮你生成所需代码，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2020-08-08
  • maven导入本地仓库jar包,报:Could not find artifact的解决

    maven导入本地仓库jar包,报:Could not find artifact的解决

    这篇文章主要介绍了maven导入本地仓库jar包，报:Could not find artifact的解决方案，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2023-03-03

最新评论