👯 👩🏼 📦 Apache Kafkaと.NET Core 2.0を使用したマイクロサービスアーキテクチャの構築 🍧 👏🏿 👨‍⚖️

良い一日！ Apache Kafkaは非常に高速な分散メッセージブローカーです。今日は、それを「調理」し、コンソールアプリケーションからシンプルなマイクロサービスアーキテクチャを実装する方法を説明します。それで、Apache Kafkaを知り、それを試してみたい方は、catにようこそ。

概要部

はじめに

この資料は、Apache Kafkaの完全な説明でも、マイクロサービスアーキテクチャの構築に関する微妙な問題でもありません。知っておく必要があるのは、.NETプラットフォームでアプリケーションを構築する方法だけです。 .Net Core 2.0を使用します

では、最終的には何を作成しますか？あなたの子供に名前を付ける方法を教えてくれるアプリケーション。簡単にするために、事前定義リストからランダムに男性と女性の名前を発行します。システムは、2つのコンソールアプリケーションと1つのライブラリで構成されます。

アイデアは、「モノリシック」ではなく、分散アプリケーションを構築することです。したがって、将来のスケーリングや、ここで説明する他の多くの利点のために予備を確保します。

システムの構造は次のとおりです。

側面にある3つの青い「長方形」はコンソールアプリケーションです。実際、以下の2つはマイクロサービスであり、MainAppはユーザーアプリケーションです。それを通じて、名前を要求します。 NameServiceは、男性または女性の名前を生成できるユニバーサルサービスになります。

中央のオレンジ色の「長方形」は、Apache Kafkaメッセージブローカーです。メッセージブローカーは、システムのすべての部分を接続するものです。今回のケースでは、Apache Kafkaを使用しますが、同じ成功を収めた場合、RabbitMQ、ActiveMQ、またはその他を使用できます。

そして、これがMainAppとApache Kafkaの相互作用です。

これは次のように機能します。

ユーザーは、いくつかのデータ（この場合は男性または女性の名前）を要求します。
MainAppは、メッセージ（図の「チーム」）をApache Kafkaに送信し、Apache Kafkaは必要なすべてのサービスを自動的に受信します。
これらのサービスは、別のメッセージ（図のデータ）をApache Kafkaに送信することで応答します。 MainAppは、必要な情報を含むApache Kafka（図では「データ」）からこのメッセージを受信し、ユーザーに提供します。

各サービスとApache Kafkaの相互作用は、同様の「双方向」スキームに従って発生します。

MainAppはNameServiceについて何も認識せず、その逆も同様であることに注意してください。すべての対話は、Apache Kafkaを介して行われます。ただし、MainAppとNameServiceの両方で同じ「通信チャネル」を使用する必要があります。実際には、これは、たとえば、MainAppがメッセージを送信するトピックの名前が、NameServiceがリッスンするトピックの名前と完全に一致する必要があることを意味します。

ご覧のとおり、この例のApache Kafkaの機能は、システムの異なる要素間でメッセージを転送することです。それが彼女の仕事であり、非常に迅速かつ確実です。もちろん、彼女には他の可能性もあります。公式ウェブサイトでそれらについて読むことができます。

Apache Kafkaとは

Apache Kafkaは分散メッセージブローカーです。実際、これはメッセージを非常に迅速かつ効率的に転送できるシステムです。 Kafkaの場合は単なるバイトシーケンスであるため、メッセージは任意のタイプのデータにできます。 Apache Kafkaは、1台のマシンまたは複数のマシンの両方で動作し、クラスターを形成してシステム全体の効率を高めます。この場合、Apache Kafkaをローカルで起動し、Apache Kafkaとの対話にはConfluentのライブラリを使用します。

Apache Kafkaの仕組みを理解することは重要です。メッセージを書き込むことができ、そこから読み取ることができます。 Kafkaのすべてのメッセージは、あるトピックまたは別のトピック（トピック）に属します。トピックはタイトルのようなものであり、Apache Kafkaに転送するメッセージごとに定義する必要があります。同様に、Kafkaからメッセージを読む場合、これらのメッセージがどのトピックに含まれるかを指定する必要があります。

このトピックはセクションに分割されており、原則としてその数を独自に示しています。トピック内のセクションの数はパフォーマンスにとって非常に重要です。これについてはここで読むことができます

実用部

Apache Kafka 0.11をダウンロードして実行する

現時点では、最新バージョンはバージョン0.11です。公式サイト（https://kafka.apache.org/downloads）からアーカイブをダウンロードし、任意のフォルダーに解凍します。さらにコンソールから、次のように2つのファイル（zookeeper-server.startおよびkafka-server-start）を実行する必要があります。

最初のコンソールを開き（ドライブCに展開した場合、万が一に備えて管理者に代わって開きます）、Kafkaでアーカイブを展開した場所に移動し、次のコマンドを入力します。

bin \ windows \ zookeeper-server-start.bat config \ zookeeper.properties

その後、すべてが正常であり、このプロセスが開始後すぐに停止しなかった場合、2番目のコンソールも開いてApache Kafka自体を開始します

bin \ windows \ kafka-server-start.bat config \ server.properties

zookeeper.propertiesとserver.propertiesでそれぞれ指定されたデフォルト設定でZookeeperとApache Kafkaを起動しました。 Zookeeperは必要な要素ですが、これがないとApache Kafkaは機能しません。

Kafkaの起動と設定に関する完全な情報は、公式Webサイトにあります。

コーディングを開始

そのため、Kafkaが起動しました。次に、「分散」アプリケーションを作成します。 2つのコンソールアプリケーションと1つのライブラリで構成されます。その結果、次のような3つのプロジェクトのソリューションが得られます。

私たちのライブラリはConfluent.Kafkaライブラリの「ラッパー」であり、Apache Kafkaと対話するために必要です。さらに、各コンソールアプリケーションで使用されます。

ライブラリは、ターゲットプラットフォームを対象としていますNET Core 2.0（ただし、プラットフォームでも同じ成功を収めることができます。NET標準）そのコードを以下に示します。そのためには、nugetパッケージConfluent.Kafkaをダウンロードする必要があります。

MessageBus.cs

using System; using System.Collections.Generic; using System.Text; using System.Threading; using Confluent.Kafka; using Confluent.Kafka.Serialization; namespace MessageBroker.Kafka.Lib { public sealed class MessageBus : IDisposable { private readonly Producer<Null, string> _producer; private Consumer<Null, string> _consumer; private readonly IDictionary<string, object> _producerConfig; private readonly IDictionary<string, object> _consumerConfig; public MessageBus() : this("localhost") { } public MessageBus(string host) { _producerConfig = new Dictionary<string, object> { { "bootstrap.servers", host } }; _consumerConfig = new Dictionary<string, object> { { "group.id", "custom-group"}, { "bootstrap.servers", host } }; _producer = new Producer<Null, string>(_producerConfig, null, new StringSerializer(Encoding.UTF8)); } public void SendMessage(string topic, string message) { _producer.ProduceAsync(topic, null, message); } public void SubscribeOnTopic<T>(string topic, Action<T> action, CancellationToken cancellationToken) where T: class { var msgBus = new MessageBus(); using (msgBus._consumer = new Consumer<Null, string>(_consumerConfig, null, new StringDeserializer(Encoding.UTF8))) { msgBus._consumer.Assign(new List<TopicPartitionOffset> { new TopicPartitionOffset(topic, 0, -1) }); while (true) { if (cancellationToken.IsCancellationRequested) break; Message<Null, string> msg; if (msgBus._consumer.Consume(out msg, TimeSpan.FromMilliseconds(10))) { action(msg.Value as T); } } } } public void Dispose() { _producer?.Dispose(); _consumer?.Dispose(); } } }

ライブラリコードの説明

ラッパーは、Apache Kafkaとのすべての対話を簡素化し、システム要素同士の対話の瞬間に焦点を当てるために作成されました。ライブラリには、SendMessage（）とSubscribeOnTopicの2つのメソッドがあり、チュートリアルのフレームワーク内にはもうありません。 SubscribeOnTopicでも、トピックをサブスクライブし、メッセージを継続的に「リッスン」します。したがって、複数のトピックをサブスクライブするには、それらを別々のストリームで実行することをお勧めします。これは、後でこのライブラリを使用してTask.Run（）構造を使用して行います。

次に、「メイン」アプリケーションMainAppを作成し、次に「マイクロサービス」NameServiceを作成します。それらのそれぞれは、男性または女性の名前を生成する責任があります。

.NET Core 2.0ターゲットプラットフォーム用のMainAppシンプルコンソールアプリケーションコードを以下に示します。その中に、先ほど構築した、MessageBroker.Kafka.Lib名前空間にあるライブラリへのリンクを追加する必要があることに注意してください。

MainApp.cs

 using System; using System.Threading; using MessageBroker.Kafka.Lib; using System.Threading.Tasks; namespace MainApp { class Program { private static readonly string bTopicNameCmd= "b_name_command"; private static readonly string gTopicNameCmd = "g_name_command"; private static readonly string bMessageReq = "get_boy_name"; private static readonly string gMessageReq= "get_girl_name"; private static readonly string bTopicNameResp = "b_name_response"; private static readonly string gTopicNameResp= "g_name_response"; private static readonly string userHelpMsg = "MainApp: Enter 'b' for a boy or 'g' for a girl, 'q' to exit"; static void Main(string[] args) { using (var msgBus = new MessageBus()) { Task.Run(() => msgBus.SubscribeOnTopic<string>(bTopicNameResp, msg => GetBoyNameHandler(msg), CancellationToken.None)); Task.Run(() => msgBus.SubscribeOnTopic<string>(gTopicNameResp, msg => GetGirlNameHandler(msg), CancellationToken.None)); string userInput; do { Console.WriteLine(userHelpMsg); userInput = Console.ReadLine(); switch (userInput) { case "b": msgBus.SendMessage(topic: bTopicNameCmd, message: bMessageReq); break; case "g": msgBus.SendMessage(topic: gTopicNameCmd, message: gMessageReq); break; case "q": break; default: Console.WriteLine($"Unknown command. {userHelpMsg}"); break; } } while (userInput != "q"); } } public static void GetBoyNameHandler(string msg) { Console.WriteLine($"Boy name {msg} is recommended"); } public static void GetGirlNameHandler(string msg) { Console.WriteLine($"Girl name {msg} is recommended"); } } }

MainAppコードのいくつかの説明

最初に多くの読み取り専用文字列変数を見ましたか？これらは、すべてのトピックの名前と、それらに送信するメッセージです。つまり、メッセージのタイトルとテキスト。トピック名が一致する必要があるため、MainAppが対話するすべてのサービスはそれらを認識する必要があります。たとえば、bTopicNameCmdは、男性の名前を取得する必要があるサービスコマンドのトピックの名前です（gTopicNameCmd-同様）。サービスは、同じ名前のトピックからサブスクライブして、メッセージを受信してから何かを実行する必要があります。

同様に、MainAppは、NameServiceサービスが有用な情報を転送するトピックにサブスクライブされます。たとえば、変数bTopicNameRespはトピックの名前で、NameServiceが生成した既製の男性の名前に提供されます。サービスはこのトピックに名前を送信し、MainAppはそこからそれらを受信します。

以下は、NameServiceのマイクロサービスコードです。ここでも、MessageBroker.Kafka.Lib名前空間で既に作成したライブラリへのリンクを追加する必要があることに注意してください

NameService.cs

 using System; using System.Threading; using System.Threading.Tasks; using MessageBroker.Kafka.Lib; namespace NameService { class Program { private static MessageBus msgBus; private static readonly string userHelpMsg = "NameService.\nEnter 'b' or 'g' to process boy or girl names respectively"; private static readonly string bTopicNameCmd = "b_name_command"; private static readonly string gTopicNameCmd = "g_name_command"; private static readonly string bTopicNameResp = "b_name_response"; private static readonly string gTopicNameResp = "g_name_response"; private static readonly string[] _boyNames = { "Arsenii", "Igor", "Kostya", "Ivan", "Dmitrii", }; private static readonly string[] _girlNames = { "Nastya", "Lena", "Ksusha", "Katya", "Olga" }; static void Main(string[] args) { bool canceled = false; Console.CancelKeyPress += (_, e) => { e.Cancel = true; canceled = true; }; using (msgBus = new MessageBus()) { Console.WriteLine(userHelpMsg); HandleUserInput(Console.ReadLine()); while (!canceled) { } } } private static void HandleUserInput(string userInput) { switch (userInput) { case "b": Task.Run(() => msgBus.SubscribeOnTopic<string>(bTopicNameCmd, (msg) => BoyNameCommandListener(msg), CancellationToken.None)); Console.WriteLine("Processing boy names"); break; case "g": Task gTask = Task.Run(() => msgBus.SubscribeOnTopic<string>(gTopicNameCmd, (msg) => GirlNameCommandListener(msg), CancellationToken.None)); Console.WriteLine("Processing girl names"); break; default: Console.WriteLine($"Unknown command. {userHelpMsg}"); HandleUserInput(Console.ReadLine()); break; } } private static void BoyNameCommandListener(string msg) { var r = new Random().Next(0, 5); var randName = _boyNames[r]; msgBus.SendMessage(bTopicNameResp, randName); Console.WriteLine($"Sending {randName}"); } private static void GirlNameCommandListener(string msg) { var r = new Random().Next(0, 5); var randName = _girlNames[r]; msgBus.SendMessage(gTopicNameResp, randName); Console.WriteLine($"Sending {randName}"); } } }

NameServiceコードの説明

このサービスは次のように機能します。

最初に、サービスが男性の名前を生成するか女性の名前を生成するかを決定します（つまり、この場合、準備されたリストからランダムな名前を選択するだけです）
対応するトピックを購読する

イベントハンドラメソッドの本文では、MainAppが既にサブスクライブしているトピックに、準備ができた名前のメッセージを送信します。そして、このイベントは、MainAppが名前を取得する必要があるというメッセージを送信するたびに発生します。

打ち上げ

理論的には、この段階で、必要なすべてのコードを備えた既成のソリューションがすでに用意されているはずです。その後、次の操作を実行できます。2つのアプリケーション（MainAppとNameService）がすぐに開始するようにソリューションを構成し、それらを開始します（ Apache Kafkaが既に実行されていることを確認してください ）。 NameServiceに「b」または「g」を入力して、男性または女性の名前を生成するようにサービスを構成し、同じ方法でMainAppの「b」または「g」を入力しますが、これらの同じ名前を取得します。次に、MainAppで名前を取得します。

この段階では、性別は1つだけです。男性のみだとします。今、私たちは女性の名前を取得したかった。 NameServiceプロジェクトが収集されたフォルダーに移動し、「 dotnet NameService.dll 」 コマンドを使用してコンソールで別のサービスを実行します。

その中に「g」コマンドを入力し、MainAppで女性の名前を要求すると、それを取得します。

ちなみに、この方法で好きなだけNameServiceエンティティを実行できます。これは、マイクロサービスアーキテクチャの利点の1つです。たとえば、サービスの1つが「落ちた」場合、システム全体がクラッシュすることはありません。まったく同じ仕事をする他のサービスがあります。

1つのこと：たとえば、NameServiceを5個実行すると、1つではなく5つの名前がMainAppに送られます。これは、server.propertiesファイルで指定されたApache Kafka設定が原因です。チュートリアルの一環として、資料を複雑にしないために、意図的にこれに触れません。

おわりに

この記事では、マイクロサービスアーキテクチャをできるだけシンプルでアクセスしやすいものにする原理を説明し、生きた例を使用して分散型Apache Kafkaメッセージブローカーを紹介しました。私はそれがうまくいくことを望み、見てくれてありがとう:)

Apache Kafkaと.NET Core 2.0を使用したマイクロサービスアーキテクチャの構築

概要部