プロデューサー（生産者）の実装

XMLによるプロデューサー設定

Spring Kafkaにおいて、メッセージ送信の中心的な役割を担うのは KafkaTemplate クラスです。まずはXMLを用いたプロデューサー側の設定手順を確認します。以下の設定では、プロデューサーファクトリとテンプレートの構築に必要なパラメータを定義しています。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans.xsd
           http://www.springframework.org/schema/context
           http://www.springframework.org/schema/context/spring-context.xsd">

    <context:property-placeholder location="classpath:kafka-config.properties" />

    <!-- プロデューサーの構成パラメータ -->
    <bean id="producerConfigMap" class="java.util.HashMap">
        <constructor-arg>
            <map>
                <entry key="bootstrap.servers" value="${kafka.broker.addresses}" />
                <entry key="retries" value="3" />
                <entry key="batch.size" value="16384" />
                <entry key="linger.ms" value="1" />
                <entry key="buffer.memory" value="33554432" />
                <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer" />
                <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer" />
            </map>
        </constructor-arg>
    </bean>

    <!-- プロデューサーファクトリの定義 -->
    <bean id="defaultProducerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
        <constructor-arg ref="producerConfigMap" />
    </bean>

    <!-- メッセージ送信用テンプレート -->
    <bean id="kafkaMsgTemplate" class="org.springframework.kafka.core.KafkaTemplate">
        <constructor-arg ref="defaultProducerFactory" />
        <property name="defaultTopic" value="system.default.topic" />
    </bean>
</beans>

上記のXMLでは、producerConfigMap でKafka接続およびシリアライゼーションに関する基本設定を行い、それを defaultProducerFactory に注入してファクトリを生成しています。最終的に kafkaMsgTemplate がこのファクトリを利用し、アプリケーションコードから簡単にメッセージを送信できる状態になります。

メッセージの送信処理

作成したテンプレートを使用して、トピック、パーティション、キーを指定してデータを送信します。現代のSpring Kafkaでは、非同期結果の処理に CompletableFuture が利用されるため、ラムダ式を用いてスマートにコールバックを登録できます。

// 指定したトピック、パーティション、キーでメッセージを送信
CompletableFuture<SendResult<String, String>> future = kafkaMsgTemplate.send(
        "user.events", 0, "user_123", "profile_updated"
);

// 送信成功時のコールバック
future.thenAccept(result -> {
    ProducerRecord<String, String> record = result.getProducerRecord();
    System.out.println("送信成功: トピック=" + record.topic() + ", パーティション=" + record.partition());
    // 成功時のビジネスロジック（例：データベースへの状態保存）
});

// 送信失敗時のコールバック
future.exceptionally(ex -> {
    System.err.println("送信失敗: " + ex.getMessage());
    // 失敗時のビジネスロジック（例：エラーログの記録、リトライキューへの登録）
    return null;
});

コンシューマー（消費者）の実装

XMLによるコンシューマー設定

続いて、メッセージを受信するコンシューマー側の設定です。コンシューマーファクトリ、リスナーコンテナの設定、そして実際のビジネスロジックを担うサービスクラスをXMLで組み立てます。

<!-- コンシューマーの構成パラメータ -->
<bean id="consumerConfigMap" class="java.util.HashMap">
    <constructor-arg>
        <map>
            <entry key="bootstrap.servers" value="${kafka.broker.addresses}" />
            <entry key="group.id" value="order-processing-group" />
            <entry key="enable.auto.commit" value="false" />
            <entry key="auto.offset.reset" value="earliest" />
            <entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer" />
            <entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer" />
        </map>
    </constructor-arg>
</bean>

<!-- コンシューマーファクトリの定義 -->
<bean id="defaultConsumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
    <constructor-arg ref="consumerConfigMap" />
</bean>

<!-- 業務用メッセージリスナーBean -->
<bean id="orderEventListener" class="com.example.messaging.OrderEventListener" />

<!-- リスナーコンテナのプロパティ設定 -->
<bean id="listenerContainerConfig" class="org.springframework.kafka.listener.ContainerProperties">
    <constructor-arg name="topics">
        <array>
            <value>order.created</value>
            <value>order.cancelled</value>
        </array>
    </constructor-arg>
    <property name="messageListener" ref="orderEventListener" />
</bean>

<!-- 並列処理用メッセージリスナーコンテナ -->
<bean id="concurrentListenerContainer" class="org.springframework.kafka.listener.ConcurrentMessageListenerContainer" init-method="doStart">
    <constructor-arg ref="defaultConsumerFactory" />
    <constructor-arg ref="listenerContainerConfig" />
    <property name="concurrency" value="3" />
</bean>

この設定により、consumerConfigMap からファクトリが生成され、listenerContainerConfig で監視対象のトピックとリスナーが紐づけられます。最後に concurrentListenerContainer がこれらを統合し、doStart メソッドによって並列消費プロセスが起動します。

メッセージの受信処理

メッセージの受信実装には、主に2つのアプローチがあります。

アプローチ1: MessageListener インターフェースの実装

インターフェースを直接実装し、onMessage メソッド内でトピックに応じたルーティングを行う古典的かつ確実な方法です。

public class OrderEventListener implements MessageListener<String, String> {
    
    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
        String topicName = record.topic();
        String payload = record.value();

        // トピック名に基づいて処理を振り分け
        switch (topicName) {
            case "order.created":
                processNewOrder(payload);
                break;
            case "order.cancelled":
                handleOrderCancellation(payload);
                break;
            default:
                System.out.println("未処理のトピック: " + topicName);
                break;
        }
    }

    private void processNewOrder(String data) {
        // 新規注文確定のビジネスロジック
    }

    private void handleOrderCancellation(String data) {
        // 注文キャンセル時のビジネスロジック
    }
}

アプローチ2: @KafkaListener アノテーションの活用

アノテーションを利用することで、トピックごとにメソッドを分離できます。SpEL式によるトピック指定や、トランザクション管理を個別に設定したい場合に特に有効です。

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Payload;

public class InventoryUpdateListener {

    @KafkaListener(topics = "${inventory.deduction.topic}", groupId = "inventory-group")
    public void handleInventoryDeduction(@Payload String stockData) {
        // 在庫控除に特化したビジネスロジック
        System.out.println("在庫データを処理中: " + stockData);
        // 必要に応じて個別のトランザクション境界を設ける
    }
}