プロデューサーとコンシューマーは獨(dú)立しているため、メッセージの処理後にコンシューマーがプロデューサーに通知する方法が必要です。一般的なアプローチは、次のような共有狀態(tài)ストアを?qū)毪工毪长趣扦埂?/p>

• 共有 ConcurrentHashMap:単一の JVM テスト環(huán)境に適しており、シンプルで効率的です。
• Redis またはその他の外部ストレージ:分散テスト環(huán)境に適しており、よりスケーラブルです。

実裝手順:

1. プロデューサー側(cè)：

   • メッセージを送信する前に、一意の ID (UUID など) を生成します。
   • ID をメッセージ ヘッダーまたはメッセージ本文の一部としてコンシューマに送信します。
   • 確認(rèn)を待つマップに ID を保存し、タイムアウトを設(shè)定します。
   • ID がマップから削除されているかどうかを定期的に確認(rèn)してください。タイムアウト後に削除されない場(chǎng)合、メッセージ処理は失敗したとみなされます。

2. 消費(fèi)者側(cè):

   • メッセージを受信したら、メッセージ內(nèi)の ID を抽出します。
   • メッセージが処理された後、ID は共有狀態(tài)ストアから削除されます。
   • Acknowledgement.acknowledge() を呼び出して、メッセージが消費(fèi)されたことを確認(rèn)します。

サンプルコード (ConcurrentHashMap を使用):

 org.springframework.kafka.core.KafkaTemplate をインポートします。
org.springframework.kafka.support.Acknowledgment をインポートします。
org.springframework.stereotype.Component をインポートします。

java.util.UUIDをインポートします。
インポートjava.util.concurrent.ConcurrentHashMap;
java.util.concurrent.TimeUnitをインポートします。

@成分
パブリック クラス MessageHandler {

    private Final KafkaTemplate<string string> kafkaTemplate;
    privatefinalConcurrentHashMap<string boolean>processedMessages = new ConcurrentHashMap();

    public MessageHandler(KafkaTemplate<string string> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(String topic, String message) throws Exception {
        文字列メッセージId = UUID.randomUUID().toString();
        処理済みメッセージ.put(メッセージId、false); // マップに追加して確認(rèn)を待機(jī)しています

        kafkaTemplate.send(トピック、メッセージID、メッセージ).get(); // メッセージが正常に送信されたことを確認(rèn)します // コンシューマの確認(rèn)を待機(jī)し、タイムアウトを設(shè)定します waitForconfirmation(messageId, 5, TimeUnit.SECONDS);
    }

    public void ConsumerMessage(String messageId, String message, Acknowledgment acknowledgment) {
        試す {
            // メッセージを処理します...
            System.out.println("消費(fèi)されたメッセージ: " message);
            処理済みメッセージ.削除(メッセージId); //確認(rèn)されたことを示す ID を削除しますacknowledgment.acknowledge();
        } catch (例外 e) {
            // 例外を処理します...
        }
    }

    private void waitForconfirmation(String messageId、長(zhǎng)いタイムアウト、TimeUnit 単位) throws Exception {
        長(zhǎng)い startTime = System.currentTimeMillis();
        while (processedMessages.containsKey(messageId)) {
            if (System.currentTimeMillis() - startTime >unit.toMillis(timeout)) {
                throw new Exception("メッセージ確認(rèn)待機(jī)中のタイムアウト: " messageId);
            }
            Thread.sleep(100); //過(guò)剰な CPU 使用率を避けるために短時(shí)間スリープします}
    }
}</string></string></string>

KafkaListener を構(gòu)成します。

 org.springframework.kafka.annotation.KafkaListener をインポートします。
org.springframework.kafka.support.Acknowledgment をインポートします。
org.springframework.stereotype.Component をインポートします。

@成分
パブリック クラス KafkaConsumer {

    プライベート最終 MessageHandler メッセージハンドラー;

    public KafkaConsumer(MessageHandler messageHandler) {
        this.messageHandler = messageHandler;
    }

    @KafkaListener(トピック = "your_topic"、groupId = "your_group_id")
    public void listen(String messageId, String message, 確認(rèn)応答) {
        messageHandler.consumeMessage(messageId, メッセージ, 確認(rèn)応答);
    }
}

注意すべき點(diǎn):

• タイムアウト メカニズム:無(wú)制限に待機(jī)することを避けるために、適切なタイムアウトを設(shè)定する必要があります。
• 例外処理:コンシューマ側(cè)では、メッセージ処理の失敗は、キューに再登録したり、エラー ログを記録したりするなど、適切に処理する必要があります。
• メッセージ ID の一意性:メッセージ ID がシステム全體で一意であることを確認(rèn)します。
• 同時(shí)実行性の問(wèn)題:複數(shù)のプロデューサーが同時(shí)にメッセージを送信する場(chǎng)合は、スレッドセーフなマップや分散ロックの使用など、同時(shí)実行性の問(wèn)題を考慮する必要があります。

要約:

Kafka 自體は、プロデューサーがコンシューマーの確認(rèn)を待つメカニズムを提供しませんが、中間狀態(tài)同期メカニズムを?qū)毪工毪长趣扦长螜C(jī)能を?qū)g現(xiàn)できます。上記のサンプル コードは、ConcurrentHashMap に基づく実裝ソリューションを提供します。 Redis やその他の外部ストレージの使用など、実際のニーズに基づいて、より適切なソリューションを選択できます。重要なのは、プロデューサーとコンシューマーの間で共有狀態(tài)を確立し、メッセージの処理ステータスを同期させることです。このようにして、統(tǒng)合テストの信頼性を効果的に向上させ、メッセージが正しく処理されることを確認(rèn)できます。

以上がSpring EmbeddedKafka プロデューサがコンシューマの確認(rèn)を待つための実裝計(jì)畫(huà)の詳細(xì)內(nèi)容です。詳細(xì)については、PHP 中國(guó)語(yǔ) Web サイトの他の関連記事を參照してください。