問題タブ [masstransit]

Masstransit をサービス バスとして使用し、アプリケーションには pub/sub パターンを使用します。

バスから加入者に届くメッセージを追跡できるかどうかを調べたいと思います。

私はjsonシリアライザを使用しています。

MT には、生のメッセージ本文を取得できる「エントリ ポイント」がありますか?

私はMassTransitを初めて使用しますが、理解できないことがあります。

すべてのノードが同じ仕事をすることができるサーバーファームがあるとしましょう。アプリケーションフレームワークはCQRSのスタイルです。つまり、公開する2つの基本的な種類のメッセージがあります。

• コマンド：サーバーの1つだけで処理する必要があります（最初のサーバーはジョブスロットが空いています）
• イベント：すべてのサーバーで処理する必要があります

非常に単純なMassTransitプロトタイプ（X秒ごとにhelloを送信するコンソールアプリケーション）を作成しました。

APIには、「公開」メソッドがあることがわかります。メッセージの種類（1台のサーバーとすべてのサーバー）を指定するにはどうすればよいですか？

「ハンドラー」構成を見ると、キューuriを指定できます。すべてのホストに同じキューを指定すると、すべてのホストにメッセージが表示されますが、実行を1つのサーバーのみに制限することはできません。

ホストの専用キューからリッスンすると、1つのサーバーだけがメッセージを処理しますが、他の種類のメッセージをブロードキャストする方法がわかりません。

私が欠けているものを理解するのを手伝ってください。

PS：気になるなら、私のメッセージングシステムはrabbitmqです。

テストするために、私はこのクラスで共通のクラスライブラリを作成しました：

また、ActualMainメソッドを実行するコンソールアプリケーションも3つあります。

MassTransitを使用して、あるプロセスからそのイベントを処理する他のすべてのプロセスにイベントを発行することを考えています。私の環境では、そのイベントに登録されている複数のプロセスが同じボックスで実行されている可能性があります。

MassTransit のサンプル コードに基づいて、マルチキャストを処理しようとしない非常に単純な概念実証を次に示します (プロセスも多くのマシンに分散されます)。以下の「メッセージ」を「イベント」と見なし、このプロセスのすべてのインスタンスでそのメッセージを取得する必要があるとします。

この C# アプリの複数のインスタンスを実行すると、メッセージを受信するインスタンスは 1 つだけですが、メッセージを受信するにはすべてのインスタンスが必要です。この動作は、キューの従来の概念と一致しているため、問題ありません。

私が知りたいのは、MSMQ/MassTransit を使用してすべてのサブスクライバーが受信するメッセージをパブリッシュできるかどうかです。これは、単一のサブスクライバーがデキューし、他のサブスクライバーが受信しないのとは対照的です。多分私はハンマーを使おうとしていて、絵筆が必要ですか？

ここで明確にするために、私はさまざまな種類のメッセージに対して 1 つのキューを共有することを考えているのではなく、MT と MSMQ を使用して、特定の種類のメッセージに対して「ピアツーピア」pub-sub をセットアップしようとしています。

消費者やサガに届く前に、各メッセージを調べたいと思います。IInboundMessageInterceptorが必要だと思いますが、カスタムを挿入する簡単な方法がわかりません。MTでメッセージインターセプトを実現するにはどうすればよいですか？および/またはカスタムIInboundMessageInterceptorを使用してバスを構成するにはどうすればよいですか？

Web プロジェクトで使用する ServiceBus 実装として MassTransit を検討しています。

Request/Responseパターンで遊んでいますが、消費者がメッセージを受信して​​応答するまでに長い遅延が発生し、要求発行者が応答を処理します。場合によっては、応答がまったく返ってこないように見えることがあります (10 分間実行したままにしておくと、まだ応答が返されません)。応答でハンドル デリゲートが呼び出されるのを見たのは、30 秒のタイムアウト期間とタイムアウト例外がスローされた後だけです。この状況では、ハンドラー デリゲートに設定されたブレークポイントがヒットします。

セットアップは標準的なものです。Web アプリがコールバックで応答を処理するために、要求を発行する Web アプリ、要求を消費して応答を送信するコンソール アプリがあります。

Castle Windsor を使用しており、コンテナーは WebActivator を使用して Web プロジェクトで初期化されます。

Web アプリ プロジェクト(ASP.NET Web API プロジェクト) では、MassTransit の WindsorInstaller は次のようになります。

コンソール アプリ プロジェクトでは、WindsorInstaller は次のようになります。

私はApiController次のGETアクションメソッドを持っています

BasicRequestCommand と BasicResponseCommand は次のようになります

そして、コンソール アプリで BasicRequestCommand に応答するハンドラー:

このすべてをローカルで実行すると、要求/応答がせいぜい数秒程度になると予想していました。構成に何か不足していますか?

さらに、MassTransit を log4net に接続したいと考えました。Windsor の log4net ロギング機能を使用しており、web.config に log4net セクションがあります。これはILogger、Windsor によって提供される実装 (および NHibernate のログ) ではすべて正常に機能しますが、これをログに使用するように MassTransit を構成する方法がドキュメントから明らかではありません。何か案は？

私は ServiceStack を使用して構築された Web API を持っています。これは、RabbitMQ 上にある MassTransit を介して TopShelf サービス バックエンドと通信します。ほとんどの場合、問題なく動作しますが、多数のリクエストがエラー キューに送られ始めるという問題が発生することがあります。

API は比較的単純で、2 つの呼び出ししかありません。私が知る限り、ログを見ると、メイン コールは正常に機能しています。ただし、もう 1 つの呼び出しは、ノードのステータスを判断するために使用される呼び出しであり、問​​題を引き起こしています。どちらの呼び出しも、Request/Respond メソッドを使用して同じ方法で実装されています。

エラーの 1 つからのログ ファイルを次に示します。

タイムアウトについて不平を言っていることに注意してください。可変長を試してみましたが、役に立たないようです。データベースへの接続を見ると、消費者が実際にそれを保持するものではないように見えるため、送信時の複数の失敗が原因でタイムアウトが発生したのか、それとも単に接続の問題なのかはわかりません. 通常、応答は 1 秒以内に発生します。

リクエストは次のように行われます。

消費者は次のとおりです。

誰かがこれで明らかに間違っていることを見ることができますか? API とサービスが異なるキューから読み取られていること (エラーは最終的に test_web_error キューに入れられる) を確認し、相関 ID と CorrelatedBy がステータス要求パケットと応答パケットに設定されていることを確認しました。前述のように、これはこの呼び出しでのみ発生し、他の呼び出しでは発生しません。これを実行している複数のサーバーもありますが、一度に1つのサーバーでのみ発生するようですが、常に同じサーバーではありません。

RabbitMQの上に直接構築する場合と比較して、 MassTransit上に構築する利点は何ですか?

MassTransit によって提供される利点の 1 つは、「タイプ」交換 (インターフェイス / タイプごとに発行、サブスクライブ) であると考えています。これにより、メッセージの内容が構造化されていないテキスト / BLOB であるプレーンな RabbitMQ 交換と比較して、メッセージの内容が構造化されます。

MassTransit によって提供されるその他の利点は何ですか?

同じエンドポイント名を使用してパブリッシャーとコンシューマーを作成すると、MassTransit がメッセージを失う状況が発生しました。

以下のコードに注意してください。コンシューマーまたはパブリッシャーのいずれかに別のエンドポイント名を使用すると (たとえば、パブリッシャーの場合は「rabbitmq://localhost/mtlossPublised」)、メッセージはパブリッシュされたものと消費されたものの両方の一致をカウントします。(サンプルのように) 同じエンドポイント名を使用すると、パブリッシュされるよりも消費されるメッセージが少なくなります。

これは予想される動作ですか？または、以下のサンプルコードを実行して、何か間違ったことをしていますか?

私は、多くの場合、正常に機能する大量輸送の物語を実装しました。ただし、メッセージがエラー キューに移動したり、消えたように見える場合があります。私はRabbitMQを使用しています。

知りたいこと: 1. メッセージがエラー キューに送られる理由/例外メッセージを取得するにはどうすればよいですか? (注: 私のハンドラー ロジックは try-catch ブロック内にあるため、これらのエラーはハンドラー ロジックが呼び出される前でも発生しているようです) 2. 失われたメッセージの原因は何ですか?

前もって感謝します。

過去6か月の学習曲線は、CQRSとDDDが主な原因であるため困難でした。

それは楽しかったです、そして私達は私達のプロジェクトの半分の道を進んでいます、そして私が掘り下げる時間がなかった領域はメッセージングフレームワークです。

現在、私はDTCを使用していないため、読み取りモデルが更新されていない場合、読み取りデータベースと書き込みデータベースの間に不整合が生じる可能性が非常に高くなります。また、私の読み取りデータベースと書き込みデータベースは同じマシン上にあります。それらを別々のマシンに置くことはないと思います。

私のシステムには大量のメッセージがないので、私の懸念はシステムの一貫性と信頼性に関係しています。

それで、NServiceBusのようなメッセージングフレームワークを配置する必要がありますか（読み取りデータベースと書き込みデータベースの両方が同じマシン上にある場合でも）、または他のオプションがありますか？はい、学習曲線はありますが、それを使用しないと、学ぶことがたくさんあると思います。

また、必要がなければレイヤーを入れたくない

考え？