とのPub/Subメッセージング NCache .NETアプリケーションを使用する
このビデオでは、 Pub/Subメッセージング NCache .NETアプリケーションを使用する場合、Pub/Subは、システム内のさまざまな部分が直接依存することなく相互に通信できるようにするメッセージングパターンです。つまり、メッセージを送信するパブリッシャーと、メッセージを受信するサブスクライバーがあります。その間には、メッセージを一方の端からもう一方の端にルーティングするブローカーがあります。
これはどこですか？ NCache が入ります。トピックを使用して、ここでブローカーとして機能します。ご覧のとおりです。このトピックは、 NCache キャッシュクラスタはメッセージを保存します。パブリッシャー情報とサブスクライバー情報を保存し、イベントをパブリッシャーまたはサブスクライバーに中継するためにも使用されます。
これらすべて、つまりこれらの仕組みによって、アプリケーション同士が互いに依存することなく、メッセージを一方の端末からもう一方の端末へ送信することが可能になるのです。
パブリッシュ/サブスクライブ型メッセージングとは何ですか？
- Publisher メッセージを送信する
- 加入者 メッセージを受信する
- ブローカー 中間に位置し、発行者から購読者へメッセージをルーティングする。
- 間接的かつ分離型 発行者と購読者。拡張性、保守性、機能拡張性に優れています。
ご用件
それでは、まず実際にトピックを作成する方法を見ていきましょう。ご覧のとおり、キャッシュ内のメッセージングサービスインターフェースを呼び出します。そして、create topic メソッドを呼び出します。もちろん、渡すのはトピック名だけです。トピックを作成すると、キャッシュ内に保持されます。トピックを削除するまで、常に存在し続けます。そして、もちろん、この上の行は、トピックを取得する方法です。トピックを作成したら、このメソッド呼び出しを使用するだけで済みます。
トピックの作成/取得
String topicName = "ExampleTopic";
ITopic topic = _cache.MessagingService.GetTopic(topicName);
if (topic == null)
{
topic = _cache.MessagingService.CreateTopic(topicName);
}メッセージの公開
出版社の概要
配送オプション
- 全て - すべての購読者にメッセージを送信します。
- どれでも - メッセージを1人の購読者にのみ送信します。
配送モード
- 同期 - 次のメッセージを送信する前に、応答を待ちます。
- 非同期 - 待機せずにバックグラウンドでメッセージを送信します。
- バルク - 複数のメッセージをまとめて一括送信します。
さて、トピックができました。次にやりたいことは、そのトピックにメッセージを投稿することです。 NCache メッセージの配信方法にはいくつかの異なる機能があります。配信オプションを選択できます。「すべて」を選択すると、登録済みの購読者全員にメッセージが送信されます。「いずれか」を選択すると、登録済みの購読者のうち1人だけにメッセージが送信されます。
さらに、配信モードを選択することで、メッセージを同期的に公開できます。非同期で送信することも可能で、その場合は当然パフォーマンスが向上します。最後に、一括送信も可能です。これは、メッセージを1つずつではなく、まとめて送信するもので、こちらも当然パフォーマンスが向上します。同期配信を選択するのは、購読者アプリケーションが、送信者から送信された順序と同じ順序でメッセージを受信するようにしたい場合です。そのため、同期配信を選択することになります。
メッセージの公開
それでは、メッセージを公開するためのコードを見ていきましょう。もちろん、まずはトピックオブジェクトであるITopicオブジェクトを取得する必要があります。次にメッセージオブジェクトを作成し、渡すパラメータはペイロードです。ここでは、注文オブジェクトをペイロードとして渡しています。メッセージの有効期限も渡すことができます。最後にpublishメソッドを呼び出します。メッセージ自体、配信オプションなどのパラメータを指定します。メッセージを非同期で公開することもできます。ここにある3番目のパラメータは、メッセージ配信失敗通知用です。これについては後ほど説明します。
ITopic topic = cache.MessagingService.GetTopic(topicName);
Message message = new Message(new Order(), TimeSpan.FromSeconds(15));
// deliver message to all subscribers
topic.Publish(message, DeliveryOption.All, true);
// deliver message to any one subscriber
topic.Publish(message, DeliveryOption.Any, true);
// send message asynchronously
topic.PublishAsync(message, deliveryOption, true);メッセージ配信失敗
そこで、あちらのオプションを有効にしてtrueに設定し、このコールバックメソッドも作成します。この場合、メッセージの配信に失敗したというログメッセージが出力されるだけです。そして、この行はイベントハンドラーを登録する方法です。前のスライドでそのパラメーターを有効にする必要があります（ここを参照）。そして、このコードを使用してイベントハンドラーを登録します。すると、サブスクライバーがメッセージを受信しないたびに、パブリッシャーに通知されます。そして、このコールバックを使用して、アプリケーションに特定の方法で反応させたり、メッセージをログに記録したりできます。
private static void MessageDeliveryFailureNotification(object sender, MessageFailedEventArgs args) {
Console.WriteLine("Failed to deliver message. " + args.MessageFailureReason);
}
// register event handler
ITopic topic = cache.MessagingService.GetTopic(topicName);
topic.MessageDeliveryFailure += MessageDeliveryFailedNotification;サブスクリプション
サブスクリプションタイプ
- 耐久性のある オフライン時にメッセージを保存し、再接続時に配信します。
- 非耐久性 ・購読者がオフラインの場合、メッセージを破棄します。
ポリシー
- 特別 - アクティブな加入者は1名のみに制限されます。
- 耐久性のある ・複数のアクティブな加入者を許可します。
配送モード
- 同期 メッセージを順序通りに配信します。
- 非同期 - メッセージは同時に配信され、配信順序は保証されません。
ご覧のように、特定のトピックにメッセージをパブリッシュするパブリッシャーが複数存在することがあります。トピック内には複数のサブスクリプションがあり、もちろん、それぞれのサブスクリプションには複数のサブスクライバーが存在します。ご覧のように、サブスクリプションにはさまざまな種類があります。永続的なものと、非永続的なものがあります。
耐久性のあるサブスクリプション
永続的なサブスクリプションの利点は、いわば永久的なサブスクリプションであるということです。たとえあなたが購読者であっても、アプリケーションがオフラインになったり、接続が切断されたりしても、メッセージは保存されます。再接続されるまで、またはメッセージの有効期限が切れるまで、メッセージは保存されます。
非永続サブスクリプション
それとは対照的に、非永続的なサブスクリプションでは、購読者アプリケーションが切断されるとメッセージが失われます。したがって、これらは一時的なサブスクリプションと考えてください。
サブスクリプションポリシー
次に、購読ポリシーを選択することもできます。「専用」では最大1人の購読者しか設定できません。「共有」では複数の購読者を設定できます。
つまり、非永続的なサブスクリプションはデフォルトで排他的になります。購読者は最大1人までしか登録できません。永続的なサブスクリプションの場合は、もちろん購読ポリシーを選択できます。わかりました。
配送モード
最後に、パブリッシャーの仕組みと同様に、サブスクリプションがメッセージを処理する際に、配信モードを同期または非同期に設定することもできます。もちろん、同期モードはメッセージの順序を最適化し、非同期モードはパフォーマンスを向上させます。
非永続的なサブスクリプションの作成
まず、非永続サブスクリプションの作成方法を見ていきましょう。ご覧のとおり、最初にトピックを作成し、次にcreate subscriptionメソッドを呼び出します。これにより、非永続サブスクリプションが作成されます。ここで渡すメソッドパラメータは、メッセージ受信コールバックのみです。メッセージ受信コールバックとは、メッセージを受信したときにサブスクライバーに通知するために使用されます。この場合は、単にメッセージを出力します。
ITopic topic = cache.MessagingService.GetTopic(topicName);
// Create and register subscribers for the given topic
// Message received callback has to be passed
ITopicSubscription subscription = topic.CreateSubscription(MessageReceivedCallback);
// Used to notify the Subscriber when it receives a message
private static void MessageReceivedCallback(object sender, MessageEventArgs args) {
Console.WriteLine("Message received for topic " + args.TopicName);
}永続的なサブスクリプションの作成
永続サブスクリプションの場合は少し異なり、もちろん複数のパラメーターを渡します。永続サブスクリプションは名前付きサブスクリプションであり、名前を付けることができる点に注意することが重要です。非永続サブスクリプションには名前を付けることはできません。永続サブスクリプションを作成するために渡す最初のパラメーターは名前で、ポリシー（共有または排他）を渡します。メッセージ受信コールバックとサブスクリプションの有効期限も指定します。
// Multiple Subscribers can subscribe to this Subscription
SubscriptionPolicy sharedPolicy = SubscriptionPolicy.Shared;
// Only one Subscriber allowed on this Subscription
SubscriptionPolicy exclusivePolicy = SubscriptionPolicy.Exclusive;
IDurableTopicSubscription subscription =
topic.CreateDurableSubscription("ExampleSubscription",
sharedPolicy,
MessageReceivedCallback,
TimeSpan.FromHours(1));メッセージをトピックにパブリッシュする
それでは、次のスライドに進む前に、私が用意したサンプルアプリケーションをお見せしたいと思います。もちろん、これは.NETアプリケーションです。ご覧のとおり、コードはすでに記述してあります。ここには4つのソリューションがあります。もちろん、サンプルデータがあります。これは注文です。注文発行者があります。プライマリ注文サブスクライバーとセカンダリ注文サブスクライバーがあります。
ご覧のとおり、私のオーダーパブリッシャーには、いくつかのメソッドがあります。このメソッドは、オーダーを非同期で公開します。もちろん、トピックオブジェクトを作成します。メッセージ配信の失敗を登録し、すべてのオーダーを反復処理してメッセージとして公開します。このメッセージの有効期限は15秒に設定されています。
そして、この場合の配信オプションは、すべての購読者に設定されています。わかりました。そして、各メッセージが配信されると、デモ目的のためだけに 5 秒の遅延を追加したことがわかります。送信後に各注文が出力されるはずです。ここには非同期オプションもあります。代わりに publish 非同期メソッドを呼び出すだけです。そして、一括パブリッシュ注文もあります。わかりました。これがメッセージ配信失敗通知です。「注文 ID の送信に失敗しました」というログメッセージが出力され、注文の詳細も送信されることがわかります。わかりました。では、購読者を実行せずにこのアプリケーションを今すぐ実行します。
しかしその前に、まずは NCache キャッシュ クラスタ。すでに demoCache という名前のキャッシュが作成されています。これは単一のサーバー ノードです。接続されているサーバーは 1 つだけで、Docker インスタンスにデプロイされています。統計情報を確認して、ここにデータがないことを確認したいだけです。完璧です。わかりました。キャッシュが起動して実行されています。これでアプリケーションを実行できます。統計情報を並べて表示できるように、これを最小限にしたいと思います。
それでは、パブリッシャーを実行してみましょう。サブスクライバーアプリケーションが実行されていないため、キャッシュにトピックが作成されます。まず、クライアントが接続されていることを確認しましょう。トピックが作成され、注文が1つずつ同期的に公開されています。トピックがキャッシュに保存されるため、キャッシュサイズも増加していることがわかります。
実際にこれらの注文をリッスンしているサブスクライバーアプリケーションがないため、パブリッシャーにメッセージの送信失敗が通知されるはずです。ご覧のとおり、メッセージの有効期限が切れたため、注文ID番号1の送信に失敗しました。5つの注文すべてについても同様の処理が行われるはずです。つまり、注文の送信に失敗しました。時間を節約するために、この件はこれでクローズします。
トピックの購読を作成する
次にデモを行う際は、まずサブスクライバーアプリケーションを実行し、次にパブリッシャーを実行します。では、その前に、こちらのコードがどのようなものかをお見せしましょう。ここで最初にキャッシュを初期化しているのがわかります。注文トピックを取得しています。注文サブスクリプションという名前で永続的なサブスクリプションを作成しています。ただし、これは私のプライベートメソッドであり、実際のAPI呼び出しではありません。
これが API 呼び出しです。わかりました。サブスクリプション名を渡しています。ポリシーは、この場合は共有だと思います。つまり、複数のサブスクライバーを持つことができます。メッセージ受信コールバックと 1 時間の時間間隔を渡しています。コールバックは単純に次のようになります。受信注文を出力し、ここに表示されているように注文の詳細を印刷します。それでは、このアプリケーションを実行します。サブスクリプションが作成されるはずです。わかりました。それが作成されたら、パブリッシャーアプリケーションを実行して、これがどのように動作するかを確認します。
これが私のサブスクライバーです。これが私のパブリッシャーです。注文ID番号1が発行され、サブスクライバーがすべての注文を受信していることがわかります。同期パブリッシングを使用しているため、サブスクライバーアプリケーションもデフォルトで同期的に処理していることに気づくでしょう。すべてが順番に受信されていることがわかります。注文ID番号5の送信に失敗したというメッセージは、前のメッセージからのものだと思います。実際に送信する前にアプリケーションをシャットダウンしたためです。そのため、今は無視します。しかし、すべてが順番に受信されていることがわかります。
それでは、メッセージを非同期で公開して、動作がどのように異なるかを見てみましょう。このメソッド呼び出しを async に変更します。パラメータは同じであるはずです。バッファの遅延を無視できるように、ここで有効期限を 60 秒に変更します。しかし、これから非同期と同期の違いをお見せします。サブスクライバー アプリケーションを実行します。すべてが閉じていることを確認してください。はい。それが起動したら、パブリッシャーを実行します。ここで期待されるのは、サブスクライバー アプリケーションがすべての注文を受信することです。ただし、順番には受信されません。すべての注文が公開され、注文を受信していることがわかりますが、注文番号 3 を受信し、次に番号 2 を受信し、次に番号 5 を受信します。最後に番号 5 と番号 1 を受信します。順番には受信されません。これが非同期と同期の違いです。
それでは、購読者側について簡単にご説明しましょう。注文処理には同期モードを選択しましたが、完全に100%機能させるには、購読者と発行者の両方で同期させる必要があります。ご覧のとおり、この設定ではデフォルトで配信モードが同期になっています。同期から非同期に変更するには、追加のパラメータとしてこれを渡す必要があります。
耐久性のあるサブスクリプション
それでは、別のことを試してみましょう。永続サブスクリプションと非永続サブスクリプションの違いを見てみましょう。まず、永続サブスクリプションの場合、アプリケーションがオフラインになってもメッセージが保持される様子をお見せします。では、まずすべてが閉じていることを確認しましょう。次に、永続サブスクリプションを作成します。最初にこのアプリケーションを実行して確認します。次に、デモのために、パブリッシングを同期に戻します。スペルミスはご容赦ください。注文を同期的にパブリッシュし、バッファ遅延を避けるために有効期限を 60 秒に変更します。では、今すぐ実行してみましょう。パブリッシャーがメッセージをパブリッシュしている間に、このサブスクライバーアプリケーションを途中でシャットダウンし、オンラインに戻して、どのように動作するかを確認します。
両方とも現在実行中です。注文番号1を受信しました。シャットダウンします。すべての注文をトピックに引き続き公開します。そして、サブスクライバーを今から戻します。新しいサブスクリプションは実際には作成されないことに注意してください。永続的であるため、現在存在するサブスクリプションを戻すだけです。アプリケーションがオフラインの間、いくつかの注文が公開され、まだそれらを受信できることがわかります。すでに5番目の注文を受信しており、3番目と4番目、おそらく2番目も受信するはずです。注文番号4も受信されました。キャッシュ統計も並べてすぐに開きます。2つのクライアントが接続されていることがわかります。1つはサブスクライバーアプリケーション、もう1つはパブリッシャーです。メッセージが実際に期限切れになる前に受信できることを願っています。そのため、時間を60秒に設定しました。もうすぐ1秒になるはずです。もちろん、メッセージが実際に期限切れにならないことを願っています。これが注文番号3です。わかりました。完璧です。
非永続サブスクリプション
それでは、ここでこれをシャットダウンして、まったく同じ状況で非永続サブスクリプションがどのように動作するかを見ていきましょう。まず、非永続サブスクリプションのコードをお見せします。注文トピックを取得していることがわかります。次に、サブスクリプション作成メソッドを呼び出し、コールバックを渡します。これにより、非永続サブスクリプションが作成されます。
ITopic topic = cache.MessagingService.GetTopic(topicName);
// Create and register subscribers for the given topic
// Message received callback has to be passed
ITopicSubscription subscription = topic.CreateSubscription(MessageReceivedCallback);
// Used to notify the Subscriber when it receives a message
private static void MessageReceivedCallback(object sender, MessageEventArgs args) {
Console.WriteLine("Message received for topic " + args.TopicName);
}では、今すぐこれを実行してみましょう。これが実行されたら、パブリッシャー アプリケーションを起動します。これで起動して実行されています。注文を受信します。これをシャットダウンして、数件の注文が公開されるまで待ってから、サブスクライバーを再実行します。では、今すぐこれを再実行しましょう。ご覧のとおり、注文番号 3 と 4 は、アプリケーションがオフラインのときに公開されたと思います。そのため、それらを受信することはできません。注文番号 5 は、起動して実行されている間に受信されました。そのため、表示されているのがわかります。しかし、明らかに、他の注文は受信できません。これらのメッセージは失われているからです。デモのために、あと 10 秒ほど待って、メッセージが保持されないことをお見せします。
注文済みメッセージ（出版社側）
それでは、その処理と並行して、メッセージの順序を確実にする方法もご説明します。先ほど説明したように、パブリッシャーがメッセージを同期的に送信するようにし、サブスクライバーを使用してメッセージを同期的に処理する必要があります。 NCache さらに、もう一つ便利な機能があります。publish メソッド呼び出し時にシーケンス名を指定することで、特定のメッセージのバッチが順番に配信されることが保証されます。では、このコードをご覧ください。メッセージを作成し、このシーケンス名でメッセージを公開しています。これをメッセージのバッチ全体に対して行います。これにより、もちろんメッセージの順序も保証されます。
ITopic topic = cache.MessagingService.GetTopic(topicName);
for (int i = 0; i < 30; i++) {
Order order = FetchAnyOrderFromDB();
Message message = new Message(order);
// Specify a unique sequence name for the messages
string sequenceName = "OrderMessages";
// Publish message with the sequence name
topic.Publish(message, DeliveryOption.All, sequenceName, true);
}注文済みメッセージ（購読者側）
そして、購読者側では、先ほども説明したように、同期配信モードか非同期配信モードかを選択するための追加パラメータとして配信モードを渡す必要があります。
ITopic topic = cache.MessagingService.GetTopic(topicName);
// Create and register subscribers for Topic
// Message received callback is specified
// DeliveryMode is set to async to ensure ordered messages
ITopicSubscription subscription = topic.CreateSubscription(MessageReceived, DeliveryMode.Sync);注文出力に戻ってみると、メッセージの有効期限が切れていることがわかります。メッセージを受信できない状態です。つまり、ほぼすべてが想定どおりに動作しています。
デモを予約
デモは以上です。このビデオを楽しんでいただけたなら幸いです。 NCache、 その 機能を使用どのようにアプリケーションに組み込むことができるかについては、お問い合わせください。 デモを予約するこの動画をご覧いただき、誠にありがとうございました。