灵活掌控Kafka消费，开启关闭，尽在掌握！

后端

2023-08-20 01:01:52

手动开启与关闭 Kafka 消费：释放系统潜能

手动控制 Kafka 消费：突破自动模式的束缚

Kafka 的自动消费模式为我们带来了无缝的数据流体验，但有时，这种不间断的消费会成为性能的阻碍。手动开启和关闭 Kafka 消费为我们提供了灵活的控制权，让系统能够根据需要调整资源分配，释放真正的性能潜力。

揭秘手动控制 Kafka 消费的优势

提升消费效率： 在资源紧张或网络状况不佳时，暂停 Kafka 消费可释放系统资源，防止数据堆积和延迟。
降低系统负载： 当系统处理能力不足时，暂停 Kafka 消费可减轻负载，避免系统崩溃或服务中断。
实现消费暂停： 在系统维护或升级期间，暂停 Kafka 消费可防止消费操作影响稳定性和数据一致性。

SpringBoot 中手动控制 Kafka 消费实战指南

导入 Kafka 依赖库： 将 Kafka 依赖项添加到你的 SpringBoot 项目中。
创建 Kafka 消费者类： 实现 KafkaConsumer 接口，创建 Kafka 消费者类。
禁用自动提交： 将 KafkaConsumer 的 enable.auto.commit 属性设置为 false。
实例化 Kafka 消费者： 在 SpringBoot 应用程序的主类中创建 Kafka 消费者实例。
开启 Kafka 消费： 调用 Kafka 消费者实例的 start() 方法启动消费。
关闭 Kafka 消费： 调用 Kafka 消费者实例的 close() 方法关闭消费。

代码示例：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
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 ManualKafkaConsumer {

    public static void main(String[] args) {
        // 消费者配置
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // 禁用自动提交

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

        // 订阅主题
        consumer.subscribe(Collections.singletonList("test-topic"));

        // 手动拉取并处理消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: " + record.value());
            }
            // 手动提交消费进度
            consumer.commitSync();
        }

        // 关闭消费者
        consumer.close();
    }
}