0Pricing
Apache Kafka & Stream Processing Fundamentals · Lección

Consumo de mensajes desde Kafka

Aprenda a crear consumidores que lean y procesen streams de datos de topics de Kafka gestionando offsets y grupos de consumidores.

Consumo de mensajes desde Kafka es una lección gratuita de Apache Kafka & Stream Processing Fundamentals en CoddyKit. Esta es la lección 2 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de Apache Kafka & Stream Processing Fundamentals, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de Apache Kafka & Stream Processing Fundamentals incluye 4 lecciones en total.

Partes de esta lección aún no han sido traducidas y se muestran en inglés.

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.

Preguntas frecuentes

¿La lección «Consumo de mensajes desde Kafka» es gratis?

Sí — el texto completo de «Consumo de mensajes desde Kafka» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de Apache Kafka & Stream Processing Fundamentals, actualiza a CoddyKit PRO. El curso de Apache Kafka & Stream Processing Fundamentals incluye 4 lecciones en total.

¿Qué aprenderé en «Consumo de mensajes desde Kafka»?

Aprenda a crear consumidores que lean y procesen streams de datos de topics de Kafka gestionando offsets y grupos de consumidores. Practicas Apache Kafka & Stream Processing Fundamentals con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar Apache Kafka & Stream Processing Fundamentals?

No se requiere experiencia previa. Apache Kafka & Stream Processing Fundamentals en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 2 de 4.

¿Cuánto tiempo toma la lección «Consumo de mensajes desde Kafka»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de Apache Kafka & Stream Processing Fundamentals?

Sí. Cada lección de Apache Kafka & Stream Processing Fundamentals incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

  1. Producción de mensajes en Kafka
  2. Consumo de mensajes desde Kafka
  3. Comprensión de particiones y offsets
  4. Claves de mensajes y estrategias de particionado
← Volver a Apache Kafka & Stream Processing Fundamentals