0Pricing
Apache Kafka & Stream Processing Fundamentals · درس

استهلاك الرسائل من Kafka

تعلّم بناء مستهلكين يقرؤون تدفقات البيانات من موضوعات Kafka ويعالجونها، مع التعامل مع الإزاحات ومجموعات المستهلكين

استهلاك الرسائل من Kafka درس مجاني في Apache Kafka & Stream Processing Fundamentals على CoddyKit. هذا هو الدرس 2 من أصل 4. يمكنك قراءة الدرس كاملاً أدناه مجاناً — ثم تمرن عليه مباشرة في المتصفح باستخدام محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7. هذا الدرس جزء من مسار التعلم في Apache Kafka & Stream Processing Fundamentals، وتقدمك يتزامن عبر الويب وتطبيق CoddyKit. تتضمن دورة Apache Kafka & Stream Processing Fundamentals 4 دروس في المجموع.

بعض أجزاء هذا الدرس لم تُترجم بعد وتظهر باللغة الإنجليزية.

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.

الأسئلة الشائعة

هل درس «استهلاك الرسائل من Kafka» مجاني؟

نعم — نص درس «استهلاك الرسائل من Kafka» كامل متاح مجاناً هنا على الويب. لتمرينه بشكل تفاعلي (محرر أكواد مدمج ومدرس ذكاء اصطناعي متاح 24/7) وفتح باقي دورة Apache Kafka & Stream Processing Fundamentals، انتقل إلى CoddyKit PRO. تتضمن دورة Apache Kafka & Stream Processing Fundamentals 4 دروس في المجموع.

ماذا ستتعلم في «استهلاك الرسائل من Kafka»؟

تعلّم بناء مستهلكين يقرؤون تدفقات البيانات من موضوعات Kafka ويعالجونها، مع التعامل مع الإزاحات ومجموعات المستهلكين تتمرن على Apache Kafka & Stream Processing Fundamentals مع أكواد عملية تشغلها مباشرة في المتصفح، ومدرس ذكاء اصطناعي متاح 24/7 يجيب على أسئلتك أثناء عملك.

هل أحتاج إلى خبرة سابقة لأبدأ Apache Kafka & Stream Processing Fundamentals؟

لا تُشترط خبرة سابقة. Apache Kafka & Stream Processing Fundamentals على CoddyKit منظم للمبتدئين حتى المتقدمين، لذا يمكنك البدء من هنا أو من البداية والتقدم بسرعتك الخاصة. هذا هو الدرس 2 من أصل 4.

كم من الوقت يستغرق درس «استهلاك الرسائل من Kafka»؟

معظم دروس CoddyKit تستغرق حوالي 5–10 دقائق. كل منها موجز وتفاعلي، لذا تحرز تقدماً مستمراً وتستأنف من حيث توقفت عبر الويب والتطبيق.

هل يمكنني كتابة وتشغيل أكواد في درس Apache Kafka & Stream Processing Fundamentals هذا؟

نعم. كل درس في Apache Kafka & Stream Processing Fundamentals يتضمن محرر أكواد مدمج، لذا تكتب وتشغل أكواداً حقيقية مباشرة في متصفحك وتحصل على تعليقات فورية من الذكاء الاصطناعي — بدون إعداد محلي.

جميع الدروس في هذه الدورة

  1. إنتاج الرسائل إلى Kafka
  2. استهلاك الرسائل من Kafka
  3. فهم الأقسام والإزاحات
  4. مفاتيح الرسائل واستراتيجيات التقسيم
← العودة إلى Apache Kafka & Stream Processing Fundamentals