Javaを使用してApache Kafkaメッセージを作成および消費(fèi)するための鍵は、プロデューサーと消費(fèi)者APIを適切に構(gòu)成し、基本的なプロセスを理解することです。 1.最初にKafkaクライアントの依存関係を追加して、バージョンがクラスターと互換性があることを確認(rèn)します。 2.プロデューサーを作成するときは、Bootstrap.Servers、Key.Serializer and Value.Serializerを構(gòu)成し、Kafkaproducerインスタンスを作成してメッセージを送信し、クロージングリソースとオプションのコールバック処理に注意を払います。 3.消費(fèi)者を書くときは、group.id、deserializerなどを構(gòu)成する場(chǎng)合、kafkaconsumerを使用してトピックをサブスクライブし、ループでメッセージを引き出し、オフセットの送信の戦略に注意してください。 4.一般的な問題には、Group.IDの競(jìng)合、オフセットの不適切な提出、および誤ったリソースの閉鎖などが含まれます。最初にローカルスタンドアロン環(huán)境で検証プロセスをテストすることをお?jiǎng)幛幛筏蓼埂?

実際、JavaでApache Kafkaメッセージを作成して消費(fèi)することは難しくありません。コアは、Kafkaの基本概念を理解し、生産者と消費(fèi)者APIの使用を習(xí)得することにあります。構(gòu)成が適切であり、コード構(gòu)造が明確である限り、メッセージの送信と受信を簡(jiǎn)単に実現(xiàn)できます。

Kafka依存関係を追加します

コードの作成を開始する前に、Kafkaのクライアントライブラリをプロジェクトに紹介する必要があります。 Mavenを使用している場(chǎng)合は、 pom.xmlに次の依存関係を追加できます。

 <依存関係>
    <groupid> org.apache.kafka </groupid>
    <artifactid> kafka-clients </artifactid>
    <バージョン> 3.6.0 </version>
</依存関係>

このバージョンは比較的新しく、比較的安定しています。もちろん、Kafkaクラスターバージョンに基づいて、対応するクライアントバージョンを選択することもできます。

Kafkaプロデューサーを書いてください

プロデューサーのタスクは、Kafkaトピックにメッセージを送信することです。主な手順には、プロパティの構(gòu)成、プロデューサーインスタンスの作成、メッセージレコードの作成、送信が含まれます。

キー構(gòu)成項(xiàng)目は次のとおりです。

• bootstrap.servers ：Kafka Brokerアドレス
• key.serializer and value.serializer ：キーと値のシリアル化方法を指定します。 StringSerializerが一般的に使用されます。

サンプルコードは次のとおりです。

 Properties Props = new Properties（）;
props.put（ "Bootstrap.servers"、 "localhost：9092"）;
props.put（ "key.serializer"、 "org.apache.kafka.common.serialization.stringserializer"）;
props.put（ "value.serializer"、 "org.apache.kafka.common.serialization.stringserializer"）;

プロデューサー<string、string> producer = new kafkaproducer <>（props）;
produceRecord <string、string> record = new produceRrecord <>（ "my-topic"、 "key"、 "value"）;

producer.send（レコード）;
producer.close（）;

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

• リソースの漏れを避けるために送信した後、 close()を呼び出すのが最善です
• 送信が成功したかどうかを確認(rèn)する場(chǎng)合は、コールバックfunction.send .send(record, callback)を追加できます。
• それが単なるテストの場(chǎng)合、キーを省略できます

Kafka Consumerを書いてください

消費(fèi)者は、Kafkaトピックからメッセージを読む責(zé)任があります。プロデューサーと比較して、消費(fèi)者のロジックは、メッセージを積極的に引っ張り、オフセットを処理する必要があるため、もう少し複雑です。

重要な構(gòu)成には次のものがあります。

• bootstrap.servers ：ブローカーアドレスも指定します
• group.id ：消費(fèi)者グループID、設(shè)定する必要があります。そうしないと、エラーが報(bào)告されます
• key.deserializer and value.deserializer ：deserializer、通常はStringDeserializerを使用します

単純な消費(fèi)者プロセスは次のとおりです。

 Properties Props = new Properties（）;
props.put（ "Bootstrap.servers"、 "localhost：9092"）;
props.put（ "group.id"、 "test-group"）;
props.put（ "key.deserializer"、 "org.apache.kafka.common.serialization.stringdeserializer"）;
props.put（ "value.deserializer"、 "org.apache.kafka.common.serialization.stringdeserializer"）;

kafkaconsumer <string、string> consumer = new Kafkaconsumer <>（props）;
consumer.subscribe（collections.singletonlist（ "my-topic"））;

while（true）{
    ConsumerRecords <String、String> Records = consumer.poll（duration.ofmillis（100））;
    for（counterrecord <string、string> record：records）{
        system.out.println（ "receiv：" record.value（））;
    }
}

注記：

• poll()メソッドはデータのバッチを引いてループで処理します
• 消費(fèi)後にオフセットを送信することを忘れないでください。自動(dòng)または手動(dòng)で送信することを選択できます
• 1回だけ消費(fèi)してから終了する場(chǎng)合は、処理後にループを破ることができます

よくある質(zhì)問と予防策

メッセージが受信されていないか、繰り返し消費(fèi)されていないことがわかります。これは通常、次の理由によって引き起こされます。

• Consumer Group.ID設(shè)定エラーまたは競(jìng)合
• オフセット周波數(shù)を自動(dòng)的に送信すると高すぎるか低すぎます
• 生産者や消費(fèi)者を正しく閉鎖できないと、一貫性のないステータスが生じます
• Kafka Brokerが起動(dòng)されていないか、ネットワークが接続されていません

開発中にローカルスタンドアロン環(huán)境を最初に実行してからクラスターに移動(dòng)することをお?jiǎng)幛幛筏蓼埂?/p>

基本的にそれだけです。 JavaでKafkaに接続するプロセスは複雑ではありませんが、シリアル化、Group.ID、および提出戦略など、いくつかの詳細(xì)は無視できます。構(gòu)成とプロセスに注意を払う限り、すぐに実行できます。

以上がJavaでApache Kafkaからメッセージを作成して消費(fèi)する方法は？の詳細(xì)內(nèi)容です。詳細(xì)については、PHP 中國(guó)語 Web サイトの他の関連記事を參照してください。