.NET で Kafka を使用する方法

[[386220]]

この記事はWeChatの公開アカウント「Coder Reading」から転載したもので、著者はCoder Readingです。この記事を転載する場合は、Coder Reading の公開アカウントにご連絡ください。

Kafka は、オープンソースで分散型、スケーラブル、高性能なパブリッシュ/サブスクライブ メッセージング ミドルウェアです。大量のデータを処理するシステムを構築したい場合、Kafka は非常に良い選択肢になります。この記事では、Kakfa に基づいてパブリッシュ/サブスクライブ プログラムを構築する方法について説明します。

Kafka アーキテクチャ

このセクションでは、まず Kafka のインフラストラクチャと関連する用語について見てみましょう。一般的に、Kafka は次のコンポーネントで構成されています。

• Kafka クラスター 1 つ以上のサーバーで構成されるクラスター
• プロデューサー メッセージを公開するために使用されるコンポーネント。
• コンシューマー メッセージを取得して処理するコンポーネント。
• ZooKeeper は、分散環境内の各ノードの構成情報を保存するのによく使用される集中型調整コンポーネントです。

Kafka では、データの基本単位はキーと値のペアであるメッセージです。 Kafkaはすべてのメッセージをbyte[]に変換します。プロデューサー、コンシューマー、クラスターはすべて TCP プロトコルを使用して通信することに注意してください。 Kafka クラスター内の各マシンはブローカーと呼ばれます。クラスターにマシンを簡単に追加して、容量の水平拡張を実現できます。

次の図は、Kafka の基本的なアーキテクチャを示しています。

Kafka のトピックはメッセージの論理的なコレクションを表します。わからない場合は、トピックをカテゴリとして考えることができます。カテゴリの下には自然に分類されたメッセージがあり、これらのメッセージはプロデューサーによって生成されます。

Kafka サーバーには 1 つ以上のトピックが含まれ、各トピックには 1 つ以上のパーティションを含めることができます。パーティションは、順序付けられたメッセージのシーケンスとして定義されます。パーティションが Kafka の動的拡張の鍵となることは注目に値します。つまり、パーティションを複数の Kafka サーバーに分散できます。具体的な操作プロセスは、Kafka のプロデューサーが指定されたトピックにメッセージをプッシュし、トピックをサブスクライブしているコンシューマーがメッセージを取得できることです。

Kafka と RabbitMQ の比較

Kafka と RabbitMQ はどちらも非常に人気のあるオープンソースのメッセージ ミドルウェアです。では、RabbitMQ ではなく Kakfa を選択すべきなのはどのような場合でしょうか?主な考慮事項は次のとおりです。

• RabbitMQ は、高性能言語 Erlang で書かれています。豊富なルーティング メカニズムと強力なメッセージ確認メカニズムを備えています。 RabbitMQ は視覚的な WebUI インターフェイスも提供しており、これを使用して RabbitMQ の実行ステータスを監視できます。ただし、大規模な展開が必要な場合、RabbitMQ は Kafka ほど適していません。Kafka はパーティションを追加することによってのみ拡張できるためです。
• RabbitMQ クラスターには典型的なスプリットブレイン問題があり、個別のプラグイン サポート (フェデレーション) を使用する必要があります。
• Kafka はパフォーマンスにおいて RabbitMQ をはるかに上回ります。単一ノードの Kafka は 1 秒あたり 100,000 件のレコードを処理できますが、RabbitMQ は 1 秒あたり約 20,000 件のレコードしか処理できません。

生産者と消費者の育成

このセクションでは、Kafka のプロデューサーとコンシューマーを構築する方法について説明します。これには、それぞれの役割を果たす 2 つのコンソール プログラムを構築する必要があります。 nuget を使用して kafka-net をインストールできます。コマンドは次のとおりです。

インストールパッケージ kafka-net

プロデューサーコンソールを構築する

静的void Main(文字列[] 引数)
        {
            文字列ペイロード = "Kafka へようこそ!" ;
            文字列トピック = "IDGTestTopic" ;
            メッセージ msg = 新しいメッセージ(ペイロード);
            URI uri = new Uri( "http://localhost:9092" );
            var options = new KafkaOptions(uri);
            var router = 新しい BrokerRouter(オプション);
            var client = 新しいプロデューサー (ルーター);
            client.SendMessageAsync(トピック、新しいリスト<Message> { msg }).Wait();
            コンソールの行を読み取ります。
        }

消費者向けコンソールを構築する

静的void Main(文字列[] 引数)
        {
            文字列トピック = "IDGTestTopic" ;
            URI uri = new Uri( "http://localhost:9092" );
            var options = new KafkaOptions(uri);
            var router = 新しい BrokerRouter(オプション);
            var コンシューマー = 新しいコンシューマー(新しい ConsumerOptions(トピック、ルーター));
            foreach (consume.Consume()内の変数メッセージ)
            {
                Console.WriteLine(Encoding.UTF8.GetString(message.Value));
            }
            コンソールの行を読み取ります。
        }

最後に、プロデューサー プログラムとコンシューマー プログラムを順番に起動すると、「Welcome to Kafka!」というメッセージが表示されます。コンシューマー コンソールに表示されます。

実際、オープンソースの世界には、RabbitMQ、MSMQ、IBM MQ Series など、メッセージ ミドルウェアが数多く存在します。現在、Kafka はメッセージ ミドルウェアであるだけでなく、ビッグ データ用のストリーミング処理プラットフォームでもあります。 Kafka は、IoT プログラム、ログ集約、その他の低レイテンシで強力なメッセージ保証のシナリオでもよく使用されます。アプリケーションに高速かつスケーラブルなメッセージ ミドルウェアが必要な場合は、Kafka が最適な選択肢となります。 Kafka に関する記事を後ほど共有します。

ネットでApache Kafkaメッセージングを使用する方法