Kafka 中的消费者偏移量跟踪

2025年5月14日 | 11 分钟阅读

理解 Kafka 中的位移

位移是唯一的数字标识符，表示 Kafka 主题分区中的特定位置。它们是跟踪消费者消费进度的基础。

关键概念

分区： Kafka 主题被划分为多个分区以进行并行处理。
位移： 分区中的每条消息都被分配一个位移。它从 0 开始并按顺序递增。
消费者位置： 消费者需要使用位移来跟踪它们已经消费了哪些消息。这可以确保消费者不会重新处理消息（除非需要）。

位移的重要性

消息顺序： 位移保留了分区内的消息顺序。
可重放性： Kafka 允许消费者通过重置位移来重新消费消息。
故障恢复： 位移帮助消费者在发生故障时从上次中断的地方继续。

示例程序

这是一个简单的 Java 示例，说明位移消费

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

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

public class OffsetConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            records.forEach(record -> {
                System.out.printf("Offset: %d, Key: %s, Value: %s%n", record.offset(), record.key(), record.value());
            });
        }
    }
}

输出

此程序显示消费者如何从主题读取消息并打印其位移。

消费者位移管理

位移管理策略

Kafka 提供了两种主要的位移管理方式

自动位移管理： Kafka 可以定期自动提交位移 (enable.auto.commit)。
手动位移管理： 开发人员可以在处理完消息后显式提交位移 (enable.auto.commit=false)。

自动位移管理

配置简单，确保定期提交。

如果消费者在处理消息之前崩溃，位移可能仍会被提交，从而有数据丢失的风险。

手动位移管理

提供在处理完消息后提交位移的控制权。

由于开发人员必须管理提交逻辑，因此稍微复杂一些。

示例程序：手动位移管理

在此示例中，在处理每条记录后手动提交位移

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

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

public class ManualOffsetConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Offset: %d, Key: %s, Value: %s%n", record.offset(), record.key(), record.value());
                
                // Commit offsets after processing each record
                consumer.commitSync();
            }
        }
    }
}

输出

此方法保证只有成功处理的消息的位移才会被提交。

消费者位移跟踪如何工作？

在 Kafka 中，消费者位移跟踪指的是系统记住消费者中断位置的能力。位移可以在 Kafka 中提交，允许消费者在发生故障或重启后从上次中断的位置继续。

位移提交机制

Kafka 将提交的位移存储在一个名为 __consumer_offsets 的特殊内部主题中。
每次消费者提交位移时，Kafka 都会将此信息写入该主题。
消费者从此主题检索其提交的位移，以知道从何处恢复处理。

提交类型

同步提交： 确保在继续处理消息之前提交位移。
异步提交： 位移在后台提交，以避免阻塞消息处理循环。

示例程序：异步位移提交

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;

import java.util.Properties;
import java.util.Collections;

public class AsyncCommitConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            records.forEach(record -> {
                System.out.printf("Offset: %d, Key: %s, Value: %s%n", record.offset(), record.key(), record.value());
            });

            // Asynchronous commit
            consumer.commitAsync();
        }
    }
}

输出

此程序演示了异步提交的非阻塞特性。

检索和监控位移

位移检索和监控对于了解消费者当前状态至关重要，尤其是在生产环境中，需要缓解故障和延迟。

检索已提交位移

消费者可以使用 Kafka Consumer API 提供的 committed() 方法检索已提交的位移。此方法获取特定分区的最后提交位移。

监控位移

Kafka 提供内置工具，例如 kafka-consumer-groups.sh 脚本，用于监控消费者组及其位移位置。通过跟踪位移滞后（最新消息与消费者最后处理的消息之间的差异），系统管理员可以检测慢速消费者。

示例：检索已提交位移

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.Properties;
	
public class RetrieveOffsets {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        TopicPartition partition = new TopicPartition("my-topic", 0);
        consumer.assign(Collections.singletonList(partition));

        long committedOffset = consumer.committed(partition).offset();
        System.out.printf("Committed Offset: %d%n", committedOffset);
    }
}

输出

此程序演示如何检索给定分区的已提交位移。

手动与自动位移管理

在 Kafka 中，位移管理对于确保消费者在发生故障或重启后能够从正确的位置恢复消息处理至关重要。有两种主要的位移管理策略：手动和自动。了解每种方法的优缺点有助于您设计高效可靠的 Kafka 消费者。

自动位移管理

默认情况下，Kafka 自动管理位移。这意味着 Kafka 将在指定间隔后自动提交消费者正在读取的每个分区的最新位移。这可以通过 enable.auto.commit 属性进行配置。

配置

props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");  // Commit offset every 5 seconds

通过自动位移管理，Kafka 定期提交位移。这很方便，因为开发人员无需显式编写代码来提交位移。

自动位移提交如何工作

启用自动提交： 当 enable.auto.commit 设置为 true 时，Kafka 将定期提交位移。
提交间隔： auto.commit.interval.ms 属性决定 Kafka 提交位移的频率。
数据丢失风险： 由于位移是定期提交的，如果消费者在消费后但在处理消息之前崩溃，Kafka 可能会提交未处理消息的位移。这可能导致数据丢失，因为消费者在重启时可能会跳过该消息。

手动位移管理

手动位移管理使消费者可以完全控制何时提交位移。在此方法中，您通过将 enable.auto.commit 设置为 false 来禁用自动位移提交，并在消息处理后使用 commitSync() 或 commitAsync() 方法提交位移。

配置

手动位移提交方法

1. 同步提交 (commitSync)

确保在继续处理下一批消息之前提交位移。
当您需要可靠性并确保不丢失任何消息时使用。它等待 Kafka 确认位移已提交。

示例

2. 异步提交 (commitAsync)

位移在后台提交，允许消费者继续处理而无需等待 Kafka 的确认。
适用于高吞吐量系统，其中同步提交位移可能会降低消费者速度。

示例

示例程序：手动位移管理

Here's an example of manual offset management using synchronous commits:
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

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

public class ManualOffsetManagement {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Offset: %d, Key: %s, Value: %s%n", 
                                      record.offset(), record.key(), record.value());
                    
                    // Process the message (e.g., save to a database)
                    
                    // Commit the offset after processing
                    consumer.commitSync();
                }
            }
        } finally {
            consumer.close();
        }
    }
}

输出

此程序演示了在处理每条消息后如何手动提交位移。

每种方法的优缺点

自动位移管理	手动位移管理
优点	优点
1. 简单 - 无需显式管理位移。	1. 对何时提交位移进行细粒度控制。
2. 适用于可以接受偶尔数据丢失的应用程序。	2. 确保只有已处理的消息的位移才会被提交。
缺点	缺点
1. 如果在处理之前提交位移，可能导致数据丢失。	1. 实现起来更复杂。
2. 无法控制何时提交位移。	2. 需要仔细管理提交以避免性能下降。

处理消费者故障和位移管理

在 Kafka 消费者应用程序中处理故障对于维护消息一致性并防止数据丢失或重复处理至关重要。适当的位移管理是此过程的关键部分。

消费者故障类型

崩溃故障： 消费者在处理消息但未提交位移之前崩溃。
网络故障： 临时网络问题可能导致断开连接，如果位移未正确提交，这可能导致消息重新处理。
应用程序级故障： 消费者应用程序中的错误可能导致消息处理不正确或错过位移提交。

故障对位移管理的影响

故障直接影响消费者恢复处理的点。例如

如果消费者在处理消息后但在提交位移之前崩溃，该消息将重新处理。
如果消费者在处理消息之前提交位移然后崩溃，它可能会跳过未处理的消息。

处理故障的策略

幂等处理： 确保您的消息处理逻辑是幂等的，这意味着多次处理同一条消息会产生相同的结果。这在消费者在崩溃后重新处理消息时特别有用。
- 示例： 如果您的消费者将记录写入数据库，请确保处理重复项（例如，通过在主键上使用唯一约束）。
带有异常处理的手动位移提交： 仅在成功处理消息后提交位移。如果消息处理期间发生错误，您可以避免提交位移并在恢复后重新处理消息。

示例程序：使用手动位移管理处理消费者故障

这是一个示例程序，演示了如何通过使用手动位移提交和异常处理来处理故障

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

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

public class FailureHandlingConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        // Process the message (e.g., save to a database)
                        System.out.printf("Processing message: key = %s, value = %s%n", record.key(), record.value());

                        // Commit the offset only if the message was processed successfully
                        consumer.commitSync();
                    } catch (Exception e) {
                        System.err.printf("Error processing message with key = %s: %s%n", record.key(), e.getMessage());
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }
}

输出

重试失败的消息

在上面的示例中，如果发生消息处理错误，位移不会提交，因此消息将在下次轮询时重新处理。您可以实现重试逻辑以限制重新处理尝试的次数，或者将失败的消息发送到死信队列以进行进一步分析。

在 Kafka Streams 中提交位移

Kafka Streams 是一个强大的流处理库，它建立在 Kafka 的核心功能之上。它允许开发人员构建能够处理、转换和分析实时数据流的应用程序。

在 Kafka Streams 中，提交位移的工作方式与标准 Kafka 消费者略有不同，因为 Kafka Streams 旨在无状态并内部处理位移管理。

Kafka Streams 中位移管理的工作原理

Kafka Streams 的位移跟踪
- Kafka Streams 自动在内部管理位移，因此开发人员通常不需要手动提交位移。
- 该框架在应用程序成功处理任务中的记录后将位移提交到 __consumer_offsets 主题。
精确一次语义
- Kafka Streams 支持精确一次语义 (EOS)，确保每条记录只处理一次，并且仅在处理完成后才提交位移。
检查点和状态存储
- Kafka Streams 使用状态存储来持久化中间处理状态。这些存储由 Kafka 主题支持，从而实现容错恢复。当发生故障时，Kafka Streams 可以从状态存储中恢复其状态并从正确的位移继续处理。
- 检查点定期创建并持久化，允许 Kafka Streams 从最新处理的位置恢复。

Kafka Streams 中的自动位移管理

默认情况下，Kafka Streams 在处理每批记录后处理位移提交。此过程涉及

从 Kafka 主题读取记录。
在一个或多个转换操作中处理记录。
提交已处理的记录及其相应的位移。

带有自动位移管理的 Kafka Streams 示例程序

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;

import java.util.Properties;

public class KafkaStreamsExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-example");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> stream = builder.stream("input-topic");

        stream.mapValues(value -> "Processed: " + value).to("output-topic");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}

输出

这个简单的 Kafka Streams 应用程序从 input-topic 读取消息，通过在值前面添加“Processed: ”来处理它们，并将它们写入 output-topic。Kafka Streams 将在完成转换后自动提交已处理记录的位移。

Kafka Streams 和精确一次语义 (EOS)

为了在 Kafka Streams 中启用精确一次语义，您可以将应用程序配置为使用事务保证。Kafka Streams 将确保

每条消息只处理一次。
处理结果和相应的位移提交是单个原子事务的一部分。

在 Kafka Streams 中启用精确一次语义的示例

props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

通过此配置，Kafka Streams 将使用 EOS 功能来确保可靠的位移管理，并避免消息重复或丢失。

Kafka Streams 中位移管理的注意事项

有状态与无状态处理
- 在有状态流处理中，位移管理更为复杂，因为它涉及持久化中间状态。Kafka Streams 通过状态存储来管理这种复杂性。
性能开销
- 启用精确一次语义会由于事务处理所需的额外协调而带来一些性能开销。
故障恢复
- Kafka Streams 通过状态存储和检查点处理故障恢复。当发生故障时，Kafka Streams 会恢复其状态并从正确的位移继续处理。

监控消费者位移滞后

位移滞后是 Kafka 系统中最重要的监控指标之一。滞后是指 Kafka 消费者当前位置与它正在读取的分区的最新位移之间的差异。监控消费者滞后有助于确保消费者跟上消息生成到主题的速度。

消费者滞后是分区的最新位移（也称为“结束位移”）与消费者针对该分区的当前提交位移之间的差异。高消费者滞后表明消费者正在落后，无法跟上传入消息的速度。

如何计算滞后

结束位移： 分区中可用的最新位移。
提交位移： 消费者已处理并提交的最后一个位移。

示例程序：使用 Kafka API 监控消费者滞后

Kafka 提供了一个 API 来以编程方式检索消费者滞后信息。下面是一个计算特定消费者组滞后的示例。

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.clients.admin.AdminClientConfig;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class LagMonitor {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        try (AdminClient adminClient = AdminClient.create(props)) {
            // Get consumer group details
            String consumerGroup = "test-group";
            ListConsumerGroupOffsetsResult offsets = adminClient.listConsumerGroupOffsets(consumerGroup);
            offsets.partitionsToOffsetAndMetadata().get().forEach((tp, offsetAndMetadata) -> {
                try {
                    // Get the latest offset for the partition
                    long endOffset = adminClient.listOffsets(
                        Collections.singletonMap(tp, OffsetSpec.latest())
                    ).all().get().get(tp).offset();

                    long consumerOffset = offsetAndMetadata.offset();
                    long lag = endOffset - consumerOffset;

                    System.out.printf("Partition: %s, Lag: %d%n", tp, lag);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

输出

此示例检索最新位移并计算消费者组中每个分区的滞后。

下一主题Kafka-broker-in-detail

Kafka 中的消费者偏移量跟踪

理解 Kafka 中的位移

关键概念

位移的重要性

消费者位移管理

自动位移管理

手动位移管理

消费者位移跟踪如何工作？

位移提交机制

提交类型

检索和监控位移

检索已提交位移

监控位移

手动与自动位移管理

自动位移管理

每种方法的优缺点

处理消费者故障和位移管理

消费者故障类型

故障对位移管理的影响

处理故障的策略

在 Kafka Streams 中提交位移

Kafka Streams 中的自动位移管理

Kafka Streams 中位移管理的注意事项

监控消费者位移滞后

如何计算滞后

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

Kafka 中的消费者偏移量跟踪

理解 Kafka 中的位移

关键概念

位移的重要性

消费者位移管理

自动位移管理

手动位移管理

消费者位移跟踪如何工作？

位移提交机制

提交类型

检索和监控位移

检索已提交位移

监控位移

手动与自动位移管理

自动位移管理

每种方法的优缺点

处理消费者故障和位移管理

消费者故障类型

故障对位移管理的影响

处理故障的策略

在 Kafka Streams 中提交位移

Kafka Streams 中的自动位移管理

Kafka Streams 中位移管理的注意事项

监控消费者位移滞后

如何计算滞后

相关帖子

使用 Kafka 和 ML 进行自动化异常检测

使用 Kafka 向 Cosmos DB 进行实时流式传输

Kafka 到 Cosmos DB 数据流中的延迟优化

使用 Kafka Connect 演进旧系统

Kafka 生产者消费者架构

Kafka 吞吐量

Apache Kafka vs RabbitMQ

使用 Kafka 在零售业中实现超个性化

Kafka API

分区策略：Kafka 主题和 Cosmos DB 容器

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器