Kafka 日志压缩

2025年5月15日 | 阅读时间 6 分钟

Kafka 是一款功能强大的开源工具，可帮助管理和处理数据流。它旨在实时处理大量数据，并用于创建即时移动和处理数据的系统。Kafka 可以管理大量的实时数据，通常用于消息传递、日志收集、流处理和事件跟踪。它最初由 LinkedIn 创建，后来成为 Apache 软件基金会的一部分。Kafka 旨在易于扩展、容错且可靠。

Kafka 概述

Kafka 充当一个高度可扩展、容错的分布式提交日志，以容错方式记录消息，并将它们复制到多个服务器以实现可靠性和高可用性。

Kafka 将数据组织成主题，每个主题又分为多个分区。每个分区都是一个顺序的、有序的日志，记录会追加到其中。Kafka 通过复制提供持久性。

Kafka 作为日志

Kafka 作为日志的概念是理解其运作方式的基础。主题中的每个分区都是一个日志——Kafka 以线性、按时间顺序追加的记录序列。Kafka 的日志可以被认为是

持久化：日志是持久的，并写入磁盘。
仅追加：新消息追加到日志的末尾。
有序：每个分区内的记录严格有序。

Kafka 中的日志如何工作？

Kafka 为日志中的每条消息分配一个唯一的偏移量，它充当标识符和排序机制。
消费者从分区读取消息，跟踪他们已读取的偏移量，Kafka 允许他们从任何偏移量开始消费。

这种设计允许 Kafka 通过在多个代理之间复制日志来实现可扩展性和容错性，每个代理都保持副本之间的一致日志。

Kafka 生产者和日志写入

生产者向 Kafka 主题发送消息。当生产者发送消息时，Kafka 将其写入主题中的一个分区，将其添加到日志中，并为其提供唯一的偏移量。生产者可以一次发送一条消息（同步）或批量发送（异步），具体取决于他们需要系统执行的速度。

生产者示例

以下是一个 Java Kafka 生产者示例，它将消息写入 Kafka 主题

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class KafkaPE {
    public static void main(String[] args) {
        // Define properties for the Kafka producer
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // Create a new Kafka producer
        Producer<String, String> producer = new KafkaProducer<>(props);

        // Send messages to the topic
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", Integer.toString(i), "Message-" + i);
            producer.send(record);
        }

        // Close the producer
        producer.close();
    }
}

输出

Kafka 消费者和日志读取

消费者从 Kafka 主题读取消息。Kafka 允许多个消费者订阅一个主题并并发处理消息。Kafka 确保每个消费者都获得自己的数据副本，这使其成为分布式系统的理想选择。每个消费者通过维护一个偏移量从特定分区读取数据，该偏移量跟踪读取的最后一条消息。

消费者示例

以下是一个 Java Kafka 消费者示例，它从 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 KafkaCE{
    public static void main(String[] args) {
        // Define properties for the Kafka consumer
        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");

        // Create a new Kafka consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // Subscribe to the topic
        consumer.subscribe(Collections.singletonList("my-topic"));

        // Continuously poll for new messages
        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());
            }
        }
    }
}

输出

Kafka Broker

Kafka 代理是负责处理客户端请求（来自生产者和消费者）和维护日志的中间件。每个代理可以托管许多主题和分区，Kafka 代理协同工作以确保数据在不同代理之间复制。

Kafka 使用主从复制来实现高可用性。每个分区都有一个领导者，它处理所有读写请求，以及追随者，它们复制数据以防止故障。如果领导者失败，其中一个追随者将被选为新的领导者。

日志压缩

Kafka 支持一项名为日志压缩的功能，该功能可确保保留给定键的最新值，而较旧的值可能会被删除。这对于使用 Kafka 维护数据最新快照的系统很有帮助。

在日志压缩主题中

具有相同键的消息被压缩，只保留每个键的最新值。
Kafka 定期删除不再需要的旧消息，确保日志保持可管理。

Kafka 作为事件溯源的持久日志

Kafka 以日志为中心的设计使其成为实现事件溯源的绝佳选择。Kafka 可用于记录和重放这些事件，从而允许系统通过重放事件日志来重建其状态。

事件溯源示例

考虑一个用户下订单的电子商务应用程序。每个订单都作为事件记录在 Kafka 中。系统状态（例如库存水平或订单状态）可以通过重放日志中的订单事件来重建。

生产者程序：向 Kafka 发送订单事件

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

public class OrdProd {

    public static void main(String[] args) {

        // Define Kafka producer properties
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // Create Kafka producer
        Producer<String, String> producer = new KafkaProducer<>(props);

        try {
            // Create an order event
            String orderId = "order-123";
            String orderStatus = "Placed";

            // Send the order event to the Kafka topic
            ProducerRecord<String, String> orderEvent = new ProducerRecord<>("orders", orderId, orderStatus);
            producer.send(orderEvent, (RecordMetadata metadata, Exception exception) -> {
                if (exception == null) {
                    System.out.printf("Order Event sent successfully. Topic: %s, Partition: %d, Offset: %d%n", 
                                       metadata.topic(), metadata.partition(), metadata.offset());
                } else {
                    System.err.println("Error sending order event: " + exception.getMessage());
                }
            });
        } finally {
            // Close the producer to release resources
            producer.close();
        }
    }
}

输出

消费者程序：从 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 OrderConsumer {

    public static void main(String[] args) {

        // Define Kafka consumer properties
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "order-consumers");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest");  // Start consuming from the earliest message

        // Create Kafka consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // Subscribe to the "orders" topic
        consumer.subscribe(Collections.singletonList("orders"));

        // Poll for new order events and process them
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    String orderId = record.key();
                    String orderStatus = record.value();

                    // Process the order event
                    System.out.printf("Received Order Event. Order ID: %s, Status: %s, Offset: %d%n", 
                                      orderId, orderStatus, record.offset());

                    if (orderStatus.equals("Placed")) {
                        // Update system state (e.g., reduce inventory)
                        System.out.println("Processing Order: " + orderId + ". Updating inventory...");
                        // Simulate inventory reduction or any business logic
                    }
                }
            }
        } finally {
            // Close the consumer to release resources
            consumer.close();
        }
    }
}

输出

用于日志处理的 Kafka Streams API

Kafka 提供 Kafka Streams API，它支持实时处理记录流。Kafka Streams 允许您构建有状态和无状态的流处理应用程序，这些应用程序消费、处理并将记录重新生成到 Kafka 主题。

流处理示例

以下是一个简单的示例，它使用 Kafka Streams 计算 Kafka 主题中每个单词的出现次数

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

import java.util.Arrays;
import java.util.Properties;

public class WordCountExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("application.id", "wordcount-app");
        props.put("bootstrap.servers", "localhost:9092");
        props.put("default.key.serde", Serdes.String().getClass().getName());
        props.put("default.value.serde", Serdes.String().getClass().getName());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> textLines = builder.stream("input-topic");
        KTable<String, Long> wordCounts = textLines
                .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
                .groupBy((key, value) -> value)
                .count();

        wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

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

输出

处理容错

Kafka 的设计本身是容错的，具有多种机制来确保数据的可靠性和持久性

复制：Kafka 在代理之间复制分区。如果一个代理失败，另一个代理可以从复制的分区提供数据。
确认 (acks)：生产者可以配置确认行为，以确保数据仅在复制到一定数量的代理后才被视为已写入。
消费者偏移量：Kafka 存储消费者偏移量，这允许消费者在发生故障后从上次中断的地方继续读取。

处理代理故障

当 Kafka 代理发生故障时，它所管理的分区的领导权会自动转移到副本。以下是一个有助于调整 Kafka 容错性的配置示例

# Ensure that each partition has 3 replicas
replication.factor=3

# Producer config to wait for acknowledgment from all replicas
acks=all

# Number of acknowledgments required from ISR (in-sync replicas)
min.insync.replicas=2

下一主题Kafka 中的系统监控和警报

Kafka 日志压缩

Kafka 概述

Kafka 作为日志

Kafka 中的日志如何工作？

Kafka 生产者和日志写入

生产者示例

Kafka 消费者和日志读取

消费者示例

Kafka Broker

日志压缩

Kafka 作为事件溯源的持久日志

事件溯源示例

用于日志处理的 Kafka Streams API

处理容错

处理代理故障

联系信息

关注我们

教程

面试题

在线编译器

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 作为日志

Kafka 中的日志如何工作？

Kafka 生产者和日志写入

生产者示例

Kafka 消费者和日志读取

消费者示例

Kafka Broker

日志压缩

Kafka 作为事件溯源的持久日志

事件溯源示例

用于日志处理的 Kafka Streams API

处理容错

处理代理故障

相关帖子

Kafka 生产者消费者架构

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

Kafka 用于个性化内容交付系统

Kafka 控制器

Kafka 重新平衡

Kafka 在增强现实 (AR) 应用程序中的作用

Kafka-to-Cosmos DB 管道中的模式演进和序列化

Kafka 中的集群迁移

Apache Kafka 中的发布-订阅消息传递

Kafka 消息压缩

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器