問題タブ [kafka-producer-api]

メッセージを kafka にパブリッシュするための Java クライアントを作成しました。コードは以下のとおりです

このコードを実行すると、次のメッセージと例外が表示されます

これは無限ループで発生し、アプリケーションがハングします... kafka ブローカーを確認したところ、トピックが作成されたことがわかりました... しかし、メッセージが表示されませんでした... しばらくこれに悩まされていました。 .. 助けてください

私はカフカ 0.8.2 を使用しています。ドキュメントが言うように：

batch.num.messagesは次を指定します。

非同期モードの使用時に 1 回のバッチで送信するメッセージの数。プロデューサは、この数のメッセージを送信する準備ができるか、または queue.buffer.max.ms に達するまで待機します。

request.required.acksは、ブローカーによるリクエストの確認を制御します。

Kafka ブローカーはこの確認応答をどのように送信するのだろうか、バッチに対して、または個々のメッセージごとに ACK を送信するのでしょうか?

だから私は通知kafkaプロデューサーをkafkaコンシューマーに送信する単純なアプリケーションを実装したいと思います。これまでのところ、文字列メッセージをプロデューサーからコンシューマーに送信することに成功しました。利用した。

そしてこちらがプロデューサー

そして、これは消費者です

そして最後に、 customserializer を使用してオブジェクトをシリアライズおよびデシリアライズしました。

誰かが問題を教えてもらえますか? これは正しい方法ですか？

最近、テスト用の Kafka クラスターをセットアップしました。私はアイテムを聞いている消費者グループを運営していますが、うまくいっているようです。コンシューマ グループの名前は ですdefault。驚いたことに、消費者グループをリストすると、空のリストが表示されます。

また、オフセットを明示的にクエリしても何も得られません。

kafka-consumer-groups.sh —new-consumerオフセットを追跡できるようにするために、コンシューマー グループを手動で作成する必要がありますか?