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 获取Html文本中的img标签下src中的内容方法
今天小编就为大家分享一篇Java 获取Html文本中的img标签下src中的内容方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2018-06-06
Java数据结构实现二维数组与稀疏数组转换详解
稀疏数组是用于优化，压缩具有以下特点的二维数组：当二维数组中的元素大部分相同，有意义的数据元素较少时，可以使用稀疏数组进行简化，节省存储空间
2021-10-10
Java项目有中多个线程如何查找死锁
这篇文章主要介绍了Java项目有中多个线程如何查找死锁，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2021-05-05
JavaWeb如何实现限制单个账号多处登录
这篇文章主要介绍了JavaWeb如何实现限制单个账号多处登录问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2024-08-08
springboot内嵌Tomcat安全漏洞修复方式
针对CVE-2020-1938漏洞,建议升级Tomcat至安全版本以避免受影响,影响版本包括：Apache Tomcat 9.x小于9.0.31、Apache Tomcat 8.x小于8.5.51、Apache Tomcat 7.x小于7.0.100及Apache Tomcat 6.x,
2024-10-10
Java数据结构学习之二叉树
今天给大家带来的是关于Java数据结构的相关知识,文章围绕着Java二叉树展开,文中有非常详细的介绍及代码示例,需要的朋友可以参考下
2021-06-06
Java并发编程之详解CyclicBarrier线程同步
在之前的文章中已经为大家介绍了java并发编程的工具：BlockingQueue接口,ArrayBlockingQueue,DelayQueue,LinkedBlockingQueue,PriorityBlockingQueue,SynchronousQueue,BlockingDeque接口,ConcurrentHashMap,CountDownLatch,本文为系列文章第十篇,需要的朋友可以参考下
2021-06-06
idea自带database连接mysql失败问题的解决办法
在IDEA⾃带的数据库连接⼯具中,可以连接MySQL数据库,但是有的时候连接出现错误,连接不上数据库,下面这篇文章主要给大家介绍了关于idea自带database连接mysql失败问题的解决办法,需要的朋友可以参考下
2023-06-06
java银行管理系统源码
这篇文章主要为大家详细介绍了java银行管理系统源码，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2019-12-12
基于MyBatis的简单使用(推荐)
下面小编就为大家带来一篇基于MyBatis的简单使用(推荐)。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-10-10