0Pricing
Apache Kafka & Stream Processing Fundamentals · レッスン

Kafkaからのメッセージ取得

Kafkaのトピックからデータストリームを読み取り処理するコンシューマーを構築し、オフセットとコンシューマーグループを扱う方法を学習します。

「Kafkaからのメッセージ取得」はCoddyKit上の無料Apache Kafka & Stream Processing Fundamentalsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これは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時間対応のAIチューター）、Apache Kafka & Stream Processing Fundamentalsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Apache Kafka & Stream Processing Fundamentalsコースには全4レッスンが含まれています。

「Kafkaからのメッセージ取得」で何を学びますか？

Kafkaのトピックからデータストリームを読み取り処理するコンシューマーを構築し、オフセットとコンシューマーグループを扱う方法を学習します。 ブラウザで直接実行するハンズオンコードでApache Kafka & Stream Processing Fundamentalsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

Apache Kafka & Stream Processing Fundamentalsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのApache Kafka & Stream Processing Fundamentalsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。

「Kafkaからのメッセージ取得」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このApache Kafka & Stream Processing Fundamentalsレッスンでコードを書いて実行できますか？

はい。すべてのApache Kafka & Stream Processing Fundamentalsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. Kafkaへのメッセージ送信
  2. Kafkaからのメッセージ取得
  3. パーティションとオフセットの理解
  4. メッセージキーとパーティショニング戦略
← Apache Kafka & Stream Processing Fundamentalsに戻る