JA
PortuguêsEspañolFrançaisTiếng ViệtDeutschEnglish한국어日本語简体中文
ホーム / プログラミング言語 / Spring Boot

Spring BootでのKafkaメッセージ送受信

Cris The Dev · 2026/10/22 · 9 分で読めます · 55

1. 概要

Apache Kafkaは、サービス間の非同期通信のためのイベントストリーミングプラットフォームです。Spring BootはSpring Kafkaライブラリを通じてKafkaをサポートしており、KafkaProducerとpoll()ループを自分で管理する必要がなくなります。

このチュートリアルでは、小さな例を構築します：注文サービスがOrderCreatedEventイベントを送信し、コンシューマーはそのイベントを受信して在庫を予約します。最後までお読みになると、設定方法、JSONメッセージの送受信、Dead Letter Topicによるエラー処理、Embedded Kafkaでのテスト方法がわかります。

2. 準備

2.1. Maven依存関係

spring-kafkaが必要です。Spring Bootはバージョンを管理しているため、バージョンを指定する必要はありません：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

<!-- テスト用 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

spring-kafka-testはテスト用にインメモリで動作するKafka brokerを提供します。AwaitilityはThread.sleep()を使わずに非同期処理の完了を待つのに役立ちます。

2.2. ローカルでKafkaを実行

最速の方法は、Apacheの公式Kafkaイメージを使用し、KRaftモード（ZooKeeperなし）で実行することです：

docker run -d --name kafka -p 9092:9092 apache/kafka:latest

brokerはlocalhost:9092をリッスンします。

3. データモデルと設定

送信するイベントはシンプルなrecordです：

package com.nonla.shop.order;

import java.math.BigDecimal;

public record OrderCreatedEvent(String orderId, String customerId, BigDecimal amount) {
}

次に、application.ymlでの設定です：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.nonla.shop.order"

各部分の意味：

  • bootstrap-serversはbrokerアドレスです。
  • producerはkeyにStringSerializer、valueにJsonSerializerを使用します。JsonSerializerはJacksonを使ってobjectをJSONに変換します。
  • acks: allはすべてのレプリカの同期が完了してからのみbrokerが確認応答するよう強制します。この方法はより安全ですが、若干遅くなります。
  • group-idはconsumer groupを特定します。同じgroup内のインスタンスはpartitionを共有します。
  • auto-offset-reset: earliestはgroupにoffsetがない場合、consumerをtopicの最初から読むように設定します。
  • spring.json.trusted.packagesはJsonDeserializerがobjectを作成できるpackageを宣言します。この行がないと、「The class … is not in the trusted packages」エラーが発生します。

3.1. トピックの自動作成

Spring Bootは自動的にKafkaAdminを設定します。このbeanはcontextで宣言されたすべてのNewTopicを作成します：

package com.nonla.shop.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    public static final String ORDERS_TOPIC = "orders";

    @Bean
    public NewTopic ordersTopic() {
        return TopicBuilder.name(ORDERS_TOPIC)
                .partitions(3)
                .replicas(1)
                .build();
    }
}

アプリケーション起動時、まだ存在しない場合、3つのpartitionを持つorders topicが作成されます。topicが既に存在する場合、KafkaAdminはスキップし、既存の設定を上書きしません。

4. KafkaTemplateでのメッセージ送信

Spring Bootはproducer設定に基づいて自動的にKafkaTemplate beanを作成します。注入するだけです：

package com.nonla.shop.order;

import com.nonla.shop.config.KafkaTopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, OrderCreatedEvent>> publish(OrderCreatedEvent event) {
        return kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("注文 {} の送信に失敗しました", event.orderId(), ex);
                        return;
                    }
                    var metadata = result.getRecordMetadata();
                    log.info("注文 {} をpartition {} offset {} に送信しました",
                            event.orderId(), metadata.partition(), metadata.offset());
                });
    }
}

注意すべき点：

  • send(topic, key, value)はorderIdをkeyとして使用します。Kafkaはkeyをハッシュしてpartitionを選択するため、同じ注文のすべてのメッセージは同じpartitionに入ります。これにより順序が保証されます。
  • Spring Kafka 3.0以降、send()は廃止されたListenableFutureではなくCompletableFuture<SendResult>を返します。
  • send()は非同期です。呼び出しはすぐに戻り、whenComplete内のコールバックはbrokerが応答した時に実行されます。

送信に成功すると、ログは以下のようになります：

INFO  c.n.s.order.OrderEventPublisher : 注文 ORD-1001 をpartition 2 offset 0 に送信しました

batchジョブなど、送信完了まで待機する必要がある場合は、publish(event).get(5, TimeUnit.SECONDS)を呼び出すことができます。高トラフィックのrequest処理スレッドではそうしないでください。各呼び出しはbroker応答を待つスレッドを保持します。

注意： データベースへの書き込みとKafka送信を同じメソッド内で行うことは、原子的なトランザクションではありません。データベースコミットが成功してもKafka送信が失敗した場合、データが不一致になります。重要なデータの場合、Transactional Outboxパターンを検討してください。

5. @KafkaListenerでのメッセージ受信

consumerは単に@KafkaListener annotationを付けたメソッドが必要です：

package com.nonla.shop.inventory;

import com.nonla.shop.config.KafkaTopicConfig;
import com.nonla.shop.order.OrderCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class OrderCreatedListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCreatedListener.class);

    private final InventoryService inventoryService;

    public OrderCreatedListener(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_TOPIC, concurrency = "3")
    public void onOrderCreated(@Payload OrderCreatedEvent event,
                               @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                               @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("注文 {} をpartition {} offset {} から受信しました", event.orderId(), partition, offset);
        inventoryService.reserve(event);
    }
}

InventoryServiceは単なる通常のserviceです：

package com.nonla.shop.inventory;

import com.nonla.shop.order.OrderCreatedEvent;
import org.springframework.stereotype.Service;

@Service
public class InventoryService {

    public void reserve(OrderCreatedEvent event) {
        if (event.amount().signum() <= 0) {
            throw new IllegalArgumentException("無効な金額です：" + event.amount());
        }
        // 在庫を予約する...
    }
}

説明：

  • topicsはリッスンするtopicを指定します。groupIdはspring.kafka.consumer.group-idから取得されるか、annotationで直接上書きできます。
  • concurrency = "3"は3つのconsumer threadを作成し、各threadはpartitionを担当します。partitionの数より大きい値は無意味です。余分なthreadは何も受け取りません。
  • @Payloadはデシリアライズ済みobjectを受け取ります。@Headerはpartitionやoffsetなどのメタデータを取得します。
  • デフォルトでは、メソッドが完了してexceptionをスローしない場合にoffsetがコミットされます（AckMode.BATCH）。そのため、アプリケーションが途中で停止した場合、メッセージは再処理されます。これはat-least-onceセマンティクスであり、ロジックはidempotentである必要があります：同じorderIdを2回処理しても2回在庫を予約することはできません。

keyやtimestamp、あるいは任意のheaderも必要な場合は、ConsumerRecord<String, OrderCreatedEvent>全体をパラメータとして受け取ることができます。

6. エラー処理とDead Letter Topic

listenerがexceptionをスローすると、デフォルトのDefaultErrorHandlerは9回連続でリトライしてからログを記録してメッセージをスキップします。実際には、エラーメッセージを別のtopicに移動して後で検査する方がよいです：

package com.nonla.shop.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

Spring Bootは自動的にこのCommonErrorHandler beanをデフォルトのcontainer factoryにアタッチします。上記の設定は以下のように機能します：

  • FixedBackOff(1000L, 2)は最大2回リトライし、各試行の間に1秒待機します。
  • リトライが終了すると、DeadLetterPublishingRecovererはメッセージをorders.DLT topicに送信します。この名前は元のtopic名に.DLTサフィックスを追加したもので、メッセージは元のpartitionを保持します。DLT topicには、元のtopic以上のpartitionが必要です。
  • IllegalArgumentExceptionはデータエラーなので、リトライしても役に立ちません。handlerはメッセージを直接DLTに移動します。

amount = -5を持つメッセージは、以下のようなログを生成します：

ERROR o.s.k.l.DefaultErrorHandler : Backoff none exhausted for orders-1@4
INFO  c.n.s.inventory.DltListener  : DLTのエラーメッセージ：ORD-1002

注意： デシリアライズ中に発生するエラー（例えば、不正なJSON）はlistenerに到達しません。処理しない場合、consumerはそのメッセージで永遠に停止する可能性があります。これを「poison pill」と呼びます。ErrorHandlingDeserializerでデシリアライザーをラップしてください：

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

この設定により、デシリアライズエラーはDefaultErrorHandlerに転送され、エラーメッセージもDLTに入ります。

7. Embedded Kafkaでのテスト

@EmbeddedKafkaはインメモリでbrokerを起動するため、テスト実行時にDockerは不要です。以下の例は、OrderEventPublisher経由でメッセージを送信し、InventoryServiceが正しくイベントを受け取ったことを確認する完全なフローをテストします。

package com.nonla.shop.order;

import com.nonla.shop.inventory.InventoryService;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

import java.math.BigDecimal;
import java.time.Duration;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.verify;

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(partitions = 3, topics = "orders")
class OrderKafkaIntegrationTest {

    @Autowired
    private OrderEventPublisher publisher;

    @MockitoBean
    private InventoryService inventoryService;

    @Test
    void whenOrderPublished_thenListenerReservesInventory() throws Exception {
        var event = new OrderCreatedEvent("ORD-1001", "CUS-42", new BigDecimal("250000"));

        var result = publisher.publish(event).get();
        assertThat(result.getRecordMetadata().topic()).isEqualTo("orders");

        var captor = ArgumentCaptor.forClass(OrderCreatedEvent.class);
        await().atMost(Duration.ofSeconds(10))
                .untilAsserted(() -> verify(inventoryService).reserve(captor.capture()));

        assertThat(captor.getValue()).isEqualTo(event);
    }
}

各部分の役割：

  • ${spring.embedded.kafka.brokers}は@EmbeddedKafkaが提供するpropertyです。これには埋め込みbrokerのアドレスが含まれており、bootstrap-serversを上書きするために使用します。
  • @MockitoBean（Spring Boot 3.4以降）は実際のInventoryServiceをmockに置き換えて、検証しやすくします。古いバージョンでは@MockBeanを使用します。
  • publish(event).get()はbroker確認応答を待つため、メッセージが確実に書き込まれたことを保証します。
  • listenerは別のthreadで実行されるため、await().untilAsserted(...)は最大10秒間のverifyを再試行します。
  • OrderCreatedEventはrecordなので、equals()は値で比較します。assertThat(...).isEqualTo(event)はJSONが正しくシリアライズおよびデシリアライズされたことを確認します。

設定内のauto-offset-reset: earliestはテストの安定性を助けます。consumerがメッセージ送信後にgroupに参加する場合でも、そのメッセージを読むことができます。

8. よくあるエラーと使うべきでない場合

よくあるエラーをいくつか説明します：

  • trusted.packagesの欠落：JsonDeserializerは未知のclassの作成を拒否します。正しくpackageを宣言してください。全コントロール下の環境でのみ"*"を使用してください。
  • producerとconsumerのpackageが異なる：JsonSerializerはデフォルトで__TypeId__ headerにクラスの完全修飾名を書き込みます。consumerサービスが同じ名前のclassを別のpackageに持っている場合、デシリアライズは失敗します。この場合、consumerでspring.json.use.type.headers: falseとspring.json.value.default.type: com.nonla.inventory.OrderCreatedEventを設定してください。
  • listenerが時間がかかる：1回のpoll()処理がmax.poll.interval.ms（デフォルト5分）を超えると、brokerはconsumerが停止したと見なしてrebalanceします。その結果、メッセージが重複処理されます。max.poll.recordsを減らすか、ロジックを最適化してください。
  • concurrencyがpartitionの数より大きい：余分なthreadは何もしません。スループットを増やすには、まずpartition数を増やす必要があります。
  • nullのkeyを送信：メッセージはpartition間で均等に分散され、エンティティごとの順序保証が失われます。

また、Kafkaがすべての場合に最良の選択肢とは限りません：

  • 同期的なrequest/responseが必要で、すぐに結果が必要な場合。RESTやgRPCを使用してください。
  • システムが小さく、トラフィックが低く、eventの履歴を保存する必要がない場合。その場合、Kafkaクラスターの運用は余分な負担であり、RabbitMQやデータベース内のqueueテーブルで十分かもしれません。
  • メッセージごとの複雑なルーティング、またはメッセージごとに異なるlatencyや優先度が必要な場合。これは従来のmessage brokerの強みです。

9. まとめ

このチュートリアルでは、JSONシリアライザーでSpring Kafkaを設定し、NewTopicでtopicを作成しました。KafkaTemplateで順序を保証するためのkeyを使用してメッセージを送信し、concurrency付きの@KafkaListenerで受信しました。また、DefaultErrorHandlerとDead Letter Topicを組み合わせてエラーを処理し、@EmbeddedKafkaとAwaitilityを使ってintegration testを書きました。

本番環境では、3つのポイントを覚えてください：listenerはidempotentである必要があり、ErrorHandlingDeserializerでデシリアライズエラーを処理し、イベント送信がデータベース内のデータと一貫していることを確認する必要があります。

spring-bootkafkaspring-kafkamessagingjava
著者のアプリAlterTurn photos into postage stamps, collected in a stamp bookiOS · Android

同じトピックの記事