🧛 🙍🏽 😫 Windows ServerのService Busで作業中の最初のレーキ 🐐 🈹 🧒🏾

Microsoftには、Windows Server用のサービスバスとしてあまり知られていないものがあります。そして、たまたまいくつかのプロジェクトで彼女と仕事をする機会がありました。その結果、他のプロジェクトよりも頻繁にプロジェクトで見つかった小さな落とし穴を収集することが判明しました。私が共有するもの。

Windows Server用Service Busとは何かの簡単な説明

これは、Windows上のWindows Azure Service Busに非常に近いMicrosoft Service Busの実装ですが、Azure自体は必要ありません。つまり、一種の非常に快適で高度なタイヤです。標準キューとその高度なトピックトピックの両方を提供でき、同じメッセージを複数の異なるサブスクリプションに送信できます。実際にはトピック/サブスクリプションのみに遭遇したため、それらについてのみ説明します。

つまり、消費者はトピックに投稿を公開します。トピックはそれらをすべてのサブスクリプションに転送します。次に、サブスクリプションは、メッセージが必要かどうかを確認し、ルールのリストと比較します（フィルター）。すべての適切なメッセージは、これらの同じサブスクリプションにサブスクライブしている顧客に送信されます。さらに、複数の顧客が同じサブスクリプションにサブスクライブしている場合、そのうちの1人だけがメッセージを受信します。すべてがかなり標準です。

最初のステップと最初のレーキ

このことの使用はどこから始まりますか？もちろん、メッセージを送受信しようとしています。

注意、以下、非生産コードが記載されています。添付コードは、テキストの機能説明図としてのみ使用することを目的としています

var messageFactory = MessagingFactory.CreateFromConnectionString(connectionString); var publisher = messageFactory.CreateTopicClient(topicName); var listener = messageFactory.CreateSubscriptionClient(topicName, subscriptionName); listener.OnMessage(message => Console.WriteLine($"Message received: {message.GetBody<string>()}")); var brokeredMessage = new BrokeredMessage("some test message"); publisher.Send(brokeredMessage);

簡単で、コンソールにメッセージが表示されます。多くのメッセージを公開して、送受信にかかる時間を大まかに推定してみましょう。

 var stopwatch = new Stopwatch(); int messagesToSent = 200; int messagesProccessed = 0; listener.OnMessage(message => { Console.WriteLine($"Message received: {message.GetBody<string>()}"); messagesProccessed++; if (messagesProccessed == messagesToSent) { stopwatch.Stop(); Console.WriteLine($"Time passed: {stopwatch.Elapsed}"); } }); stopwatch.Start(); for (var i = 0; i < messagesToSent; i ++) { var brokeredMessage = new BrokeredMessage($"Message №{i}"); publisher.Send(brokeredMessage); }

このコードを実行すると、古いベテランのコンピューターでは、処理に約6秒かかります。

しかし、次のステップはしばしば最初のレーキにつながります。実際には、サブスクライバーは次の2つのモードのいずれかでメッセージを受信できます。

PeekLock-メッセージは受信されますが、サブスクリプションから削除されず、ロックのみを受信します。削除するためには、クライアントはCommit（）を呼び出して処理の成功を明示的に確認する必要があります。それ以外の場合、ロックの有効期限が切れるか、 Abandon（）を呼び出すことにより、クライアントはこのメッセージを再度配信しようとします。
ReceiveAndDelete-メッセージが受信され、サブスクリプションからすぐに削除されます。処理が失敗した場合、このメッセージは再び表示されません。ただし、ロックがハングしないため、 PeekLockよりも少し速く動作します。

デフォルトでは、 messageFactory.CreateSubscriptionClientはPeekLockオプションを作成します。しかし、その非自明性のため、クライアントが動作モードの明示的な指示なしに作成されたことを実際には見ませんでした。また、ドキュメントによると、 PeekLockが指定されている場合、メッセージごとに.Complete（）を呼び出す必要があります。これを試してみましょう：

 listener.OnMessage(message => { Console.WriteLine($"Message received: {message.GetBody<string>()}"); messagesProccessed++; if (messagesProccessed == messagesToSent) { stopwatch.Stop(); Console.WriteLine($"Time passed: {stopwatch.Elapsed}"); } message.Complete(); //  ,        });

そして、予期しないことが起こります。実行がスローされないという事実にもかかわらず、「メッセージ番号X」の行は実行されますが、すべてが非常にゆっくりと発生します。これらの200のメッセージは、6秒ではなく、4分9秒もかかりました。これは古い鉄を正当化するものではありません。しかし、ライブプロジェクトのコードでこの問題を一度発見しました。少数のメッセージについては、パフォーマンスの低下は明らかではありませんでした。

なぜこれが起こっているのですか？結局のところ、何かが正しくない場合、実行が期待されるでしょうか？実際、受付はそうです。まったく明確な理由ではないが、Microsoftはこれらの例外に関する情報を取得するための極めて明白な方法を作成しました。

OnMessageメッセージサブスクリプションメソッドは、オプションのOnMessageOptionsパラメーターを受け入れます。これにより、 ExceptionReceivedイベントをサブスクライブできます。つまり、同じ「隠された例外」です。

 var onMessageOptions = new OnMessageOptions(); onMessageOptions.ExceptionReceived += (sender, args) => Console.WriteLine($"Exception received: {args.Exception}"); listener.OnMessage(message => { Console.WriteLine($"Message received: {message.GetBody<string>()}"); messagesProccessed++; if (messagesProccessed == messagesToSent) { stopwatch.Stop(); Console.WriteLine($"Time passed: {stopwatch.Elapsed}"); } message.Complete(); }, onMessageOptions); //    onMessageOptions

このコードを実行すると、 Microsoft.ServiceBus.Messaging.MessageLockLostExceptionが各メッセージでスローされることがわかります。

指定されたロックは無効です。彼女は期限切れになっているか、メッセージが既にキューから削除されています.. TrackingId：54630ae4-6e4f-4979-8fc8-b66e5314079c_GAPC_BAPC、TimeStamp：08.24.2016 21:20:08

なぜこれが起こっているのですか？ onMessageOptionsにはもう1つのパラメーターAutoCommitがあるためです。そして、デフォルトはtrueです。したがって、メッセージのライフサイクルを独立して制御する場合、正しく動作するには、このフィールドをfalseに設定する必要があります。これを試してみましょう：

 var stopwatch = new Stopwatch(); int messagesToSent = 200; int messagesProccessed = 0; var onMessageOptions = new OnMessageOptions { AutoComplete = false //   }; onMessageOptions.ExceptionReceived += (sender, args) => Console.WriteLine($"Exception received: {args.Exception}"); listener.OnMessage(message => { Console.WriteLine($"Message received: {message.GetBody<string>()}"); messagesProccessed++; if (messagesProccessed == messagesToSent) { stopwatch.Stop(); Console.WriteLine($"Time passed: {stopwatch.Elapsed}"); } message.Complete(); }, onMessageOptions); stopwatch.Start(); for (var i = 0; i < messagesToSent; i ++) { var brokeredMessage = new BrokeredMessage($"Message №{i}"); publisher.Send(brokeredMessage); }

出来上がり：実行なし、メッセージ処理にかかる時間はわずか2.5秒です。通常の操作のように見えます。

要約すると ：

ほとんどの場合、 onMessageOptions.ExceptionReceivedをサブスクライブする必要があります。そうしないと、コード内の多くの問題に気付かない場合があります。
デフォルトでは、サービスベースはメッセージをコミットしようとすることに注意してください。ほとんどの場合、この動作はオフにする必要があります。

抽象化とレーキ2

実稼働コードでも見られる2番目の、それほど一般的ではない瞬間は、サブスクライバーのラッパーの誤った作成です。一般的に、サービスベースを操作する内部に隠れているクラスを作成することは良いことです。しかし、微妙な違いがあります。これを行わない方法の例を示しますが、実際には同様のコードが複数回見られています。

この種類のクラスは作成されます：

 class Listener : IListener { private readonly MessagingFactory _messageFactory; private readonly SubscriptionClient _client; public event Func<string, Task> OnReceivedAsync; public Listener(string connectionString, string topicName, string subscriptionName, ReceiveMode receiveMode) { _messageFactory = MessagingFactory.CreateFromConnectionString(connectionString); _client = _messageFactory.CreateSubscriptionClient(topicName, subscriptionName, receiveMode); var onMessageOptions = new OnMessageOptions { AutoComplete = false }; onMessageOptions.ExceptionReceived += (sender, args) => Console.WriteLine($"Exception received: {args.Exception}"); _client.OnMessageAsync(bm => OnReceivedAsync?.Invoke(bm.GetBody<string>()), onMessageOptions); } }

さらに次のように使用されます：

 var messageFactory = MessagingFactory.CreateFromConnectionString(connectionString); var publisher = messageFactory.CreateTopicClient(topicName); int messagesToSent = 20; for (var i = 0; i < messagesToSent; i++) { var brokeredMessage = new BrokeredMessage($"Message №{i}"); publisher.Send(brokeredMessage); } var listener = new Listener(connectionString, topicName, subscriptionName, ReceiveMode.ReceiveAndDelete); listener.OnReceivedAsync += x => { Console.WriteLine($"Message received: {x}"); return Task.FromResult(true); };

このコードを実行すると、すべてが機能しているように見えますが、最初のメッセージの代わりに「 NullReferenceException：Object reference not set to object of instance。 」というエラーが表示されます。

さらに、onMessageOptions.ExceptionReceivedにサブスクライブした場合にのみエラーがキャッチされます。これを行わない場合（および何らかの理由で頻繁に行わない場合）、コード動作の間接的かつ非常にわかりにくいバグによってのみ問題を見つけることができます。

ここで何が間違っていますか？答えはかなり明白です。もし私がそれほど頻繁に会わなければ、おそらく言及しなかったでしょう。リスナー抽象化コンストラクターで_client.OnMessageAsyncが呼び出されると、サブスクライバーは既にメッセージの受信を開始しています。そのため、それらの多く（コンストラクターとlistener.OnReceivedAsyncへのサブスクリプションがどの程度離れているかに応じて）はスキップされ、空のOnReceivedAsync？.Invokeに該当し 、論理的にnullを返します。したがって、 NullReferenceException 。

それをどうしますか？最も簡単なことは、次のようにインスタンスの作成とサブスクリプションを設定することです。

 class Listener : IListener { private readonly MessagingFactory _messageFactory; private readonly SubscriptionClient _client; public Listener(string connectionString, string topicName, string subscriptionName, ReceiveMode receiveMode) { _messageFactory = MessagingFactory.CreateFromConnectionString(connectionString); _client = _messageFactory.CreateSubscriptionClient(topicName, subscriptionName, receiveMode); } public void Subscribe(Func<string, Task> handleMessage) { var onMessageOptions = new OnMessageOptions { AutoComplete = false }; onMessageOptions.ExceptionReceived += (sender, args) => Console.WriteLine($"Exception received: {args.Exception}"); _client.OnMessageAsync(bm => handleMessage(bm.GetBody<string>()), onMessageOptions); } }

そして、このようなものをサブスクライブします：

 var listener = new Listener(connectionString, topicName, subscriptionName, ReceiveMode.ReceiveAndDelete); listener.Subscribe(x => { Console.WriteLine($"Message received: {x}"); return Task.FromResult(true); });

クラスの作成時にメッセージが失われることはなくなりました。

要約すると ：

onMessageOptions.ExceptionReceivedへのサブスクライブは、現在よりも関連性があります
client.OnMessageAsyncを呼び出した直後にメッセージの受信が開始されることを忘れないでください。抽象化を設計するときはこれを考慮してください。

すくい番号3

サブスクライバーにはすばらしいClose（）メソッドがあります。しかし、彼の行動は完全に予測可能な推測ではありません。ここでそのようなコードを実行してみましょう。メッセージの前半を送信した後、このClose（）を呼び出し、別のサブスクライバーインスタンスを通じて後半のメッセージを受信します。

 var messageFactory = MessagingFactory.CreateFromConnectionString(connectionString); var publisher = messageFactory.CreateTopicClient(topicName); var listener1 = messageFactory.CreateSubscriptionClient(topicName, subscriptionName, ReceiveMode.ReceiveAndDelete); var listener2 = messageFactory.CreateSubscriptionClient(topicName, subscriptionName, ReceiveMode.ReceiveAndDelete); int messagesToSent = 10; int messagesProccessed = 0; var onMessageOptions = new OnMessageOptions { AutoComplete = false }; onMessageOptions.ExceptionReceived += (sender, args) => Console.WriteLine($"Exception received: {args.Exception}"); listener1.OnMessage(message => { Console.WriteLine($"listener1: message received: {message.GetBody<string>()}, listener1 is closed: {listener1.IsClosed}"); messagesProccessed++; }, onMessageOptions); for (var i = 0; i < messagesToSent; i++) { var brokeredMessage = new BrokeredMessage($"Message №{i}"); publisher.Send(brokeredMessage); Thread.Sleep(50); if (i == 4) { Console.WriteLine("Closing listener1"); listener1.Close(); } } listener2.OnMessage(message => { Console.WriteLine($"listener2: message received : {message.GetBody<string>()}, listener2 is closed: {listener2.IsClosed}"); messagesProccessed++; }, onMessageOptions);

しかし、コンソールの結果は次のようになります。

listener1：受信したメッセージ：メッセージ番号0、listener1は閉じています：False

listener1：受信したメッセージ：メッセージ番号1、listener1は閉じています：False

listener1：受信したメッセージ：メッセージ番号2、listener1が閉じている：False

listener1：受信したメッセージ：メッセージNo. 3、listener1が閉じている：False

listener1：受信したメッセージ：メッセージ番号4、listener1が閉じている：False

リスナー1を閉じる

listener1：受信したメッセージ：メッセージ番号5、listener1が閉じている：True

listener2：受信したメッセージ：メッセージ番号6、listener2は閉じています：False

listener2：受信したメッセージ：メッセージ番号7、listener2は閉じています：False

listener2：受信したメッセージ：メッセージNo. 8、listener2は閉じています：False

listener2：受信したメッセージ：メッセージ番号9、listener2は閉じています：False

明らかではないでしょう？同じことを行うが、 ReceiveAndDeleteではなくPeekLock動作モードの場合、.Complete（）がSystem.OperationCanceledExceptionをスローすることを除いて、結果は同様になります：このメッセージングエンティティは既に閉じられ、中止され、破棄されました 。また、メッセージハンドラでエラーをキャッチしてAbandon（）を手動で実行すると、 Abandon（）自体がエラーをスローします。そして、これらのアクションはどちらも普通であり、 OnMessageOptionsの中に隠れていません。

そして、 ReceedAndDeleteとは対照的に、欠落したメッセージ自体は、再送信が発生したときに引き続き処理されます。

完全な出力とコンソール出力を備えたコード

 var messageFactory = MessagingFactory.CreateFromConnectionString(connectionString); var messageFactory1 = MessagingFactory.CreateFromConnectionString(connectionString); var messageFactory2 = MessagingFactory.CreateFromConnectionString(connectionString); var publisher = messageFactory.CreateTopicClient(topicName); var listener1 = messageFactory1.CreateSubscriptionClient(topicName, subscriptionName, ReceiveMode.PeekLock); var listener2 = messageFactory2.CreateSubscriptionClient(topicName, subscriptionName, ReceiveMode.PeekLock); int messagesToSent = 10; int messagesProccessed = 0; var onMessageOptions = new OnMessageOptions { AutoComplete = false }; onMessageOptions.ExceptionReceived += (sender, args) => Console.WriteLine($"Exception received: {args.Exception}"); listener1.OnMessage(message => { try { Console.WriteLine($"listener1: message received: {message.GetBody<string>()}, listener1 is closed: {listener1.IsClosed}"); messagesProccessed++; message.Complete(); } catch (Exception ex1) { Console.WriteLine($"listener1 Complete() exception: {ex1.Message}"); try { message.Abandon(); } catch (Exception ex2) { Console.WriteLine($"listener1 Abandon() exception: {ex2.Message}"); } } }, onMessageOptions); for (var i = 0; i < messagesToSent; i++) { var brokeredMessage = new BrokeredMessage($"Message №{i}"); publisher.Send(brokeredMessage); Thread.Sleep(50); if (i == 4) { Console.WriteLine("Closing listener1"); listener1.Close(); } } listener2.OnMessage(message => { Console.WriteLine($"listener2: message received : {message.GetBody<string>()}, listener2 is closed: {listener2.IsClosed}"); messagesProccessed++; message.Complete(); }, onMessageOptions);

listener1：受信したメッセージ：メッセージ番号0、listener1は閉じています：False

listener1：受信したメッセージ：メッセージ番号1、listener1は閉じています：False

listener1：受信したメッセージ：メッセージ番号2、listener1が閉じている：False

listener1：受信したメッセージ：メッセージNo. 3、listener1が閉じている：False

listener1：受信したメッセージ：メッセージ番号4、listener1が閉じている：False

リスナー1を閉じる

listener1：受信したメッセージ：メッセージ番号5、listener1が閉じている：True

listener1 Complete（）例外：このメッセージングエンティティは既に閉じられているか、中止されているか、破棄されています。

listener1 Abandon（）例外：このメッセージングエンティティは既に閉じられているか、中止されているか、破棄されています。

listener2：受信したメッセージ：メッセージ番号6、listener2は閉じています：False

listener2：受信したメッセージ：メッセージ番号7、listener2は閉じています：False

listener2：受信したメッセージ：メッセージNo. 8、listener2は閉じています：False

listener2：受信したメッセージ：メッセージ番号9、listener2は閉じています：False

listener2：受信したメッセージ：メッセージ番号5、listener2は閉じています：False

それで何をするのか、どうやって生きるのか？これについては、コードでこれを覚えて考慮する必要があります。すべてを知っているstackoverflowは、この動作に対処するのに十分なオプションを提供します。たとえば、必要に応じて、サブスクライバーを閉じるのと同時にmessageFactory.Close（）を呼び出すことができます。または、サブスクライバーがif（listener.IsClosed）{/ *** /}などのように閉じられている場合はハンドラーをチェックインします。

要約すると ：

サービスベースからのすべてのekspressnyがonMessageOptions.ExceptionReceivedでのみ到着するわけではありません。
サブスクライバーとの作業を完了するとき、 Close（）メソッドの機能を考慮する必要があります

おわりに

一般に、Windows Server用のService Busは非常に優れた機能であり、適切に機能しますが、最初のいくつかのささいなことにはいくらかの血が流れます。この記事に記載されているポイントが誰かに役立つことを証明し、彼ら自身のコーンを詰め込むことから救うことを願っています。

Windows ServerのService Busで作業中の最初のレーキ

最初のステップと最初のレーキ

抽象化とレーキ2

すくい番号3

おわりに

More articles: