Kafkaコンシューマーの構築
Kafkaトピックを購読し、メッセージを処理するSpring Kafkaコンシューマーを開発します。
「Kafkaコンシューマーの構築」はCoddyKit上の無料Spring Boot 4 Microservices & REST APIsレッスンです。 これはレッスン2/3です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはSpring Boot 4 Microservices & REST APIs学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 Spring Boot 4 Microservices & REST APIsコースには全3レッスンが含まれています。
このレッスンの一部はまだ翻訳されておらず、英語で表示されています。
Kafka Consumers: The Listeners
In event-driven architectures, Kafka Consumers are the components responsible for reading messages (records) from Kafka topics. Think of them as listeners waiting for new events!
They subscribe to one or more topics and process the incoming data, enabling different parts of your application or other services to react to events.
Spring Boot & Kafka Config
To build a Kafka consumer in Spring Boot, first, you need the spring-kafka dependency. Add it to your pom.xml:
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>Next, configure your Kafka broker details in application.properties. This tells your Spring Boot app where to find the Kafka server.
spring.kafka.bootstrap-servers=localhost:9092Meet @KafkaListener
Spring for Apache Kafka provides the powerful @KafkaListener annotation. This annotation marks a method to be a Kafka listener, meaning it will automatically consume messages from specified topics.
topics: The Kafka topic(s) to listen to.groupId: Identifies the consumer group. Essential for scaling.
It handles all the low-level Kafka API details for you!
Your First Kafka Listener
Let's create a simple consumer that listens to a topic named my-first-topic and prints any incoming string messages to the console.
Notice the groupId. All consumers with the same groupId are part of a consumer group.
package com.coddykit.kafka;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@SpringBootApplication
@EnableKafka
public class KafkaConsumerApp {
public static void main(String[] args) {
SpringApplication.run(KafkaConsumerApp.class, args);
}
}
@Component
class SimpleKafkaListener {
@KafkaListener(topics = "my-first-topic", groupId = "my-group-id")
public void listen(String message) {
System.out.println("Received Message: " + message);
}
}Understanding Deserialization
Kafka messages are stored as byte arrays. When a consumer reads a message, it needs to convert these bytes back into a usable object (like a String or a custom Java object).
This process is called deserialization. You configure deserializers in application.properties:
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializerThe choice of deserializer depends on how the producer serialized the message.
Consumer Groups for Scale
Consumer groups are key to Kafka's scalability. Multiple consumer instances can belong to the same group, sharing the workload of consuming messages from a topic.
- Each message in a topic partition is delivered to only one consumer instance within a group.
- If you have more consumers than partitions, some consumers will be idle.
- If a consumer fails, another consumer in the same group automatically takes over its partitions.
This allows for both high availability and horizontal scaling.
Listening for JSON Data
Often, you'll send complex data as JSON. To consume JSON, you'll need to define a Java class (POJO) that matches the JSON structure and use Spring Kafka's JsonDeserializer.
Add spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer to your config.
package com.coddykit.kafka;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
// Define a simple DTO matching the JSON structure
class MyEvent {
private String name;
private int value;
// Default constructor required for deserialization
public MyEvent() {}
public MyEvent(String name, int value) {
this.name = name;
this.value = value;
}
public String getName() { return name; }
public void setName(String name) { this.name = name; }
public int getValue() { return value; }
public void setValue(int value) { this.value = value; }
@Override
public String toString() {
return "MyEvent{" +
"name='" + name + '\'' +
", value=" + value +
'}';
}
}
@SpringBootApplication
@EnableKafka
public class KafkaJsonConsumerApp {
public static void main(String[] args) {
SpringApplication.run(KafkaJsonConsumerApp.class, args);
}
}
@Component
class JsonKafkaListener {
@KafkaListener(topics = "my-json-topic", groupId = "json-group")
public void listenJson(MyEvent event) {
System.out.println("Received JSON Event: " + event);
}
}Graceful Error Handling
What happens if a message is malformed or your processing logic throws an error? Consumers need robust error handling.
For simple errors, a try-catch block within your listener method is effective. For more advanced scenarios, Spring Kafka offers error handlers:
DefaultErrorHandler: Retries messages with backoff.DeadLetterPublishingRecoverer: Sends failed messages to a dead-letter topic.
These prevent a single bad message from stopping your entire consumer.
Peeking at Message Metadata
Sometimes, you need more than just the message payload. Kafka messages come with useful metadata, such as the topic name, partition, and offset.
You can access this metadata directly in your @KafkaListener method using annotations like @Header or by accepting a ConsumerRecord object.
@KafkaListener(topics = "my-topic", groupId = "my-group")
public void listenWithInfo(
String message,
@Header(org.springframework.kafka.support.KafkaHeaders.RECEIVED_TOPIC) String topic,
@Header(org.springframework.kafka.support.KafkaHeaders.RECEIVED_PARTITION_ID) int partition
) {
System.out.println("From topic " + topic + ", partition " + partition + ": " + message);
}Quick Check: Kafka Consumers
Which of the following is the primary annotation used in Spring Kafka to mark a method as a message listener for a specific topic?
Recap: Building Kafka Consumers
Great job! You've learned how to build Spring Kafka consumers to process messages from topics.
- We set up Spring Kafka and used
@KafkaListenerto create message-consuming methods. - We explored deserialization and how to consume both simple strings and complex JSON objects.
- You also understand the importance of consumer groups for scaling and handling errors.
Next up, we'll dive deeper into integrating producers and consumers to build full event-driven microservices!
よくある質問
「Kafkaコンシューマーの構築」レッスンは無料ですか？
はい。「Kafkaコンシューマーの構築」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、Spring Boot 4 Microservices & REST APIsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Spring Boot 4 Microservices & REST APIsコースには全3レッスンが含まれています。
「Kafkaコンシューマーの構築」で何を学びますか？
Kafkaトピックを購読し、メッセージを処理するSpring Kafkaコンシューマーを開発します。 ブラウザで直接実行するハンズオンコードでSpring Boot 4 Microservices & REST APIsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
Spring Boot 4 Microservices & REST APIsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのSpring Boot 4 Microservices & REST APIsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/3です。
「Kafkaコンシューマーの構築」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このSpring Boot 4 Microservices & REST APIsレッスンでコードを書いて実行できますか？
はい。すべてのSpring Boot 4 Microservices & REST APIsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- Kafkaプロデューサー入門
- Kafkaコンシューマーの構築
- イベント駆動マイクロサービス統合