0Pricing
Apache Kafka & Stream Processing Fundamentals · Lektion

Nachrichten aus Kafka konsumieren

Lernen Sie, Consumer zu entwickeln, die Datenströme aus Kafka-Topics lesen und verarbeiten und dabei Offsets und Consumer-Gruppen verwalten.

Nachrichten aus Kafka konsumieren ist eine kostenlose Apache Kafka & Stream Processing Fundamentals-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des Apache Kafka & Stream Processing Fundamentals-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der Apache Kafka & Stream Processing Fundamentals-Kurs umfasst insgesamt 4 Lektionen.

Teile dieser Lektion wurden noch nicht übersetzt und werden auf Englisch angezeigt.

What are Kafka Consumers?

Kafka Consumers are applications that read data from Kafka topics. They subscribe to one or more topics and process the messages as they arrive.

  • Consumers are the 'listeners' in your Kafka ecosystem.
  • They pull data from Kafka brokers, rather than having data pushed to them.
  • Essential for building real-time data pipelines and analytics systems.

Working with Consumer Groups

Consumers often work together in a consumer group. This is a collection of consumer instances that share a common group.id.

  • Scalability: Messages from a topic's partitions are distributed among consumers in the same group, allowing parallel processing.
  • Fault Tolerance: If a consumer instance fails, its assigned partitions are automatically reassigned to other active consumers in the group.
  • Each message in a partition is delivered to only one consumer within a group.

How Consumers Read Data (Polling)

Kafka consumers don't just 'receive' messages. Instead, they actively poll Kafka brokers for new data.

  • The consumer repeatedly asks Kafka, "Do you have any new records for me?"
  • This polling mechanism gives the consumer control over its processing rate.
  • If no new records are available, the poll() method will simply return an empty list after waiting for a specified duration.

Basic Java Consumer Setup

To create a Kafka consumer in Java, you use the KafkaConsumer class. First, you need to set up some basic properties like the Kafka server address and deserializers.

Here's the start of a simple consumer program:

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class SimpleConsumerSetup {
    public static void main(String[] args) {
        // 1. Define consumer properties
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-first-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 2. Create the consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

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

        // Consumer loop and closing will go here
        System.out.println("Consumer setup complete. Subscribed to 'my-topic'.");
        consumer.close();
    }
}

Essential Consumer Properties

Let's break down the key properties we set for our consumer:

  • BOOTSTRAP_SERVERS_CONFIG: A list of Kafka broker addresses (e.g., localhost:9092) that the consumer will connect to.
  • GROUP_ID_CONFIG: A unique identifier for the consumer group this consumer belongs to (e.g., my-first-group).
  • KEY_DESERIALIZER_CLASS_CONFIG: Specifies how to convert message keys from bytes (received from Kafka) into Java objects (e.g., StringDeserializer for text keys).
  • VALUE_DESERIALIZER_CLASS_CONFIG: Specifies how to convert message values from bytes into Java objects (e.g., StringDeserializer for text values).

The Consumer Loop: Polling for Records

Consumers continuously poll Kafka for new data. This is typically done in an infinite loop. The poll(Duration timeout) method returns a batch of ConsumerRecords or an empty list if no new records are available within the specified timeout.

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class PollingConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-polling-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

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

        try {
            while (true) { // Infinite loop to continuously poll
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) {
                    System.out.println("No records received. Waiting...");
                } else {
                    System.out.println("Received " + records.count() + " records.");
                    // Processing of records would happen here
                }
            }
        } finally {
            consumer.close();
            System.out.println("Consumer closed.");
        }
    }
}

Processing Received Messages

Once you receive ConsumerRecords from poll(), you can iterate over them to process each message individually. Each ConsumerRecord contains the topic, partition, offset, key, and value of the message.

This is where your application's specific logic for handling the data lives.

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class ProcessingConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-processing-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        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("Topic: %s, Partition: %d, Offset: %d, Key: %s, Value: %s%n",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value());
                    // Add your custom processing logic here, e.g., store in database, perform analytics
                }
            }
        } finally {
            consumer.close();
            System.out.println("Consumer closed.");
        }
    }
}

Tracking Progress with Offsets

An offset is a unique, sequential ID assigned to each message within a Kafka topic partition. It acts like a pointer, indicating the position of a message.

  • Consumers use offsets to track which messages they have already processed.
  • Kafka stores the latest committed offset for each consumer group and partition.
  • This ensures that if a consumer restarts or a partition is reassigned, it knows exactly where to resume reading to avoid reprocessing or missing messages.

Offsets are crucial for reliable message processing.

Automatic Offset Committing

By default, Kafka consumers are configured to auto-commit offsets periodically. This is convenient for simple applications but might lead to message loss or duplicates in certain failure scenarios.

  • Set enable.auto.commit to true (this is the default setting).
  • The auto.commit.interval.ms property determines how often offsets are committed (e.g., every 5 seconds).
  • This approach is generally suitable for 'at-least-once' processing where some duplicates are acceptable.

Manual Offset Control

For more precise control over message processing and stronger guarantees (like 'exactly-once' processing), you can manually commit offsets after your application has successfully processed a batch of messages.

  • Set enable.auto.commit to false to disable automatic commits.
  • Use consumer.commitSync(): This method blocks until the commit operation is successful.
  • Use consumer.commitAsync(): This method is non-blocking and faster, but requires a callback function for handling potential errors.

Manual commits ensure that records are processed *before* their offsets are marked as 'done', preventing data loss on consumer failure.

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class ManualCommitConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-manual-commit-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // Disable auto-commit

        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("Processed: Partition %d, Offset %d, Value: %s%n",
                        record.partition(), record.offset(), record.value());
                    // Simulate processing work that must complete before committing
                }
                if (!records.isEmpty()) {
                    consumer.commitSync(); // Commit offsets after processing the batch
                    System.out.println("Offsets committed manually.");
                }
            }
        } finally {
            consumer.close();
            System.out.println("Consumer closed.");
        }
    }
}

Check Your Understanding

Consider a Kafka topic with 3 partitions. A consumer group with 2 consumer instances is subscribed to this topic. If one consumer instance fails, what happens to the partitions it was processing, and how does Kafka ensure no data is lost?

Lesson Recap: Consuming Data

Great job! In this lesson, you learned how Kafka consumers read data:

  • Consumers subscribe to topics and actively poll Kafka for new messages.
  • Consumer groups enable scalable and fault-tolerant message processing across multiple consumer instances.
  • Offsets are sequential IDs that track a consumer's progress within a topic partition.
  • You can choose between automatic (simpler, default) or manual (more control, better guarantees) offset committing for reliability.

Next, we'll dive deeper into partitions and offsets to understand how Kafka achieves high throughput and message ordering.

Häufig gestellte Fragen

Ist die Lektion „Nachrichten aus Kafka konsumieren“ kostenlos?

Ja — der vollständige Text von „Nachrichten aus Kafka konsumieren“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des Apache Kafka & Stream Processing Fundamentals-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der Apache Kafka & Stream Processing Fundamentals-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Nachrichten aus Kafka konsumieren“?

Lernen Sie, Consumer zu entwickeln, die Datenströme aus Kafka-Topics lesen und verarbeiten und dabei Offsets und Consumer-Gruppen verwalten. Du übst Apache Kafka & Stream Processing Fundamentals mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um Apache Kafka & Stream Processing Fundamentals zu starten?

Keine Vorkenntnisse erforderlich. Apache Kafka & Stream Processing Fundamentals auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.

Wie lange dauert die Lektion „Nachrichten aus Kafka konsumieren“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser Apache Kafka & Stream Processing Fundamentals-Lektion Code schreiben und ausführen?

Ja. Jede Apache Kafka & Stream Processing Fundamentals-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Nachrichten an Kafka senden
  2. Nachrichten aus Kafka konsumieren
  3. Partitionen und Offsets verstehen
  4. Nachrichtenschlüssel und Partitionierungsstrategien
← Zurück zu Apache Kafka & Stream Processing Fundamentals