問題タブ [apache-kafka]

Java プログラムで kafka クラスをインポートしようとしていますが、次のようなインポートする必要がある Java クラスが見つかりませんでした。

これらの Java クラスがどこにあり、どのように提供すればよいか誰か教えてもらえますか? Kafka は Scala で書かれているように見えるので、ソース コードをダウンロードした後、それらの Java クラスを見つけることができません。

ロギングに log4j を使用する Storm クラスターを実行しています。標準のファイルベースのロギングに加えて、kafka アペンダーを追加したいと考えています。

私のlog4j構成ファイルは次のようになります。

意図したとおりに機能するファイル アペンダーと定義がまったく同じであるにもかかわらず、kafka アペンダーが定義された ConversionPattern を使用しないことを除いて、すべて正常に機能します。kafka アペンダーを機能させるには、どのように構成を変更する必要がありますか?

誰でも知っていますか

kafka.javaapi.producer.Producer

いくつかのメソッド呼び出し (たとえば、いくつかの send(...)) の間で再利用できますか、それとも毎回閉じる必要がありますか?

Github（または他の場所）でApache Kafka（バージョン0.8が推奨）を使用した優れたデモプロジェクトはありますか？私たちはいくつかのおもちゃのプロジェクトでそれをテストしてきました。しかし、私はいくつかの実際のより大きなプロジェクトをチェックしたいと思います。

Kafka 0.72 では、zk.connect または broker.list のいずれかを指定できることがわかりました。しかし Kafka 0.8 では、broker.list しか指定できず、broker.list を指定せずに zk.connect を指定することはできません。この場合、生産者と飼育係のバランスを取ることはできないと思います。誰かがKafka 0.8を使用している場合、またはそれについて理解していますか? どうもありがとう！

Kafka の公式ドキュメントによると、プロデューサーのスループットは約 50MB/S です。しかし、私はいくつかのテストを行います.プロデューサー全体は約2MB/Sです. テスト環境はドキュメントと同じです。1 つのプロデューサー、1 つのブローカー、1 つの Zookeeper が独立したマシンにあります。メッセージ サイズは 100 バイト、バッチ サイズは 200、フラッシュ間隔は 600 メッセージです。テスト環境は同じで、構成も同じです。では、なぜ私のテスト結果とドキュメントに記載されている内容に大きなギャップがあるのでしょうか?

kafka 0.7 では、プロデューサーに zk.connect を指定できるため、プロデューサーはブローカーの追加と削除を見つけることができます。しかし、kafka 0.8 では、プロデューサーに zk.connect を指定できません。kafka 0.8 のプロデューサーはそれを見つけることができますか? そうでない場合、システムのスケーラビリティは 0.7 バージョンよりも悪くありませんか?

Kafka 0.8 では、プロデューサーがブローカーに直接接続し、Zookeeper に接続しないことがわかっています。リクエストごとに飼育係ベースの負荷分散をどのように達成するよりも? 実際、トピックが作成されると、そのパーティションは 1 つ以上のブローカーに分散されます。メッセージが送信されると、そのキーワードに従って特定のパーティションに配信されます。つまり、あるものを固定ブローカーの固定パーティションに送信する必要があります。いわゆる負荷分散はどのように機能しますか?

Apache Kafka をダウンロードし、クイックスタートを正常に完了しました。提供されたサンプル プログラムを実行しようとすると、エラーが発生しました。

メイン クラスを読み込めません。

何が欠けているのかわからないので、クラスパスを設定しました。

apache kafkaからのメッセージを消費するために、春の統合でカスタム受信チャネル アダプターを実装しようとしています。春の統合の例に基づいて、MessageSource インターフェイスを実装するクラスを作成し、kafka から消費されたメッセージを返す receive() メソッドを実装する必要があることがわかりました。ただし、 kafka のコンシューマーの例に基づいて、KafkaStream のメッセージ イテレーターは BlockingQueue によって支えられています。そのため、キューにメッセージがない場合、スレッドはブロックされます。

このメソッドは、消費するものがあるまでブロックされる可能性があるため、 receive() メソッドを実装する最良の方法は何ですか?

より一般的な意味では、消費する準備ができるまでブロックするストリーミング メッセージ ソース用のカスタム インバウンド チャネルを実装するにはどうすればよいでしょうか?