問題タブ [node-amqp]

これらのテクノロジーを統合するために何が合理的かを理解しようとしています。NodeJS (現在は amqplib を使用していますが、これは変更される可能性があります) を RabbitMQ から EasyNetQ に統合するにはどうすればよいですか?

EasyNetQがオブジェクトを期待していること（私は思う）とNode / amqplibが文字列しか送信できないことを除いて、私はそれをある程度機能させています。

C# コード:

NodeJS コード:

結果：

バスを介してオブジェクトを送信する方法はありませんか? この2つをどのように統合しますか？

私のアプリは、RabbitMQ キューを使用してメッセージを保存し、ワーカーがそれらのメッセージを消費してデータベースに挿入します。意図は、ワークロードのピーク時にデータベースに負荷をかけないことです。私が抱えている問題は、それらのピーク時にキューの発行率が非常に高くなり、ワーカーがクラッシュするまで処理できるよりも多くのメッセージを毎秒受信し始めることです。

消費率を制御して、ワーカーがメッセージを消費するよりも速くメッセージを受信しないようにする方法はありますか? メッセージは重要ではないので、ワーカが処理できるようになるまでメッセージがキューに入れられたままになる時間は気にしません。

私は Node.JS に amqplib を使用しています。これはワーカーに使用しているコードです。

コンシューマーがメッセージを受け取った後、コンシューマー/ワーカーはいくつかの検証を行ってから Web サービスを呼び出します。このフェーズでは、エラーが発生したり検証が失敗した場合、メッセージを最初に消費されたキューに戻します。

RabbitMQ のドキュメントを読みました。しかし、reject、nack、cancel メソッドの違いについて混乱しています。

現在、ユーザーが自分の Web サイトでページをロードするたびに、新しい RabbitMQ 接続を作成する必要があります。

これにより、毎回新しい TCP 接続が作成されます。ただし、NodeJS AMQP プラグインを使用して、Rabbit に対して行う TCP 接続の数を減らそうとしています。これが私が持っているものです。

私が実行している問題は、私がやろうとすると:

その後、ready 関数はトリガーされません。これは、ready コールバックが既に定義されていて、再トリガーされないためだと思います。私がやろうとしているのは、次のようにして新しいキューをバインドすることです:

この問題を回避する方法についてのご意見をお待ちしております。

ありがとう。

このライブラリを介して予期しないチャネル閉鎖に対処するための推奨される方法を探しています。

イベントが発行されていることをドキュメントから収集しましたが、そのイベントを検出する最善の方法はあまり明確ではありません (これは簡単なことですが、一般的にイベントの発行についてはよくわかりませんので、ご了承ください)。 、さらに、私がそれで何をすべきかは明らかではありません。

どのサブスクライバーがそのチャネルを使用しているかを追跡してから、再サブスクライブする必要がありますか? 他の人は何をしますか？

さらに、Channel (または ConfirmedChannel) オブジェクトを調べて、それがまだ「良い」かどうかを判断する方法はありませんか? これは、イベント トラップ アプローチよりも好ましいように思われますが、それを行う方法を見つけることができないようです (わかりました。正確にはそうではありません。チャネルの「受け入れ」メソッドを調べて、いくつかのことを行いました。悪くなっているかどうかを判断しますが、これはハックのようです)。

任意のガイダンスをいただければ幸いです。

同じキューからのメッセージを処理する必要がある 2 つのコンシューマーがありますが、常にそのうちの 1 つだけです。私が達成しようとしていることのシーケンスは次のようなものです：

1. (開始) どのコンシューマもキューにサブスクライブしていません
2. Consumer1 がキューにサブスクライブする
3. プロデューサがメッセージをキューに送信し、メッセージがコンシューマ 1 に配信されます
4. Consumer1 はメッセージを処理し、しばらくしてから購読を解除します
5. プロデューサはより多くのメッセージをキューに送信し、メッセージはキューに保存されます (autoDelete=false であるため、コンシューマがサブスクライブしていない場合でもキューは破棄されません)。
6. Consumer2 はキューにサブスクライブし、保存されたメッセージを処理し、しばらくしてからサブスクライブを解除します。
7. Consumer1 はサブスクライブし、メッセージを処理します... .. など

これは、最初は期待どおりに機能します。上記のステップ 5 の後、1 つだけがサブスクライブし、もう 1 つがサブスクライブを解除したにもかかわらず、プロデューサーからのメッセージが両方の consumer に交互に配信されることがわかります。

これを機能させるために使用しているコードは次のとおりです。

このコード、または前述の目的のシーケンスを達成するための全体的なアプローチに何か問題がありますか?

node-amqp で Express を使用しています。私の目標は、amqpConnection を作成し、サーバーが起動する前にそれをグローバル オブジェクトに保存し、Express ルートで以前に作成した を使用することですglobals.amqp_connection。

問題はconnection.on "ready"、ルートを呼び出すたびにイベントが発生することです。これは、HTTP 要求を処理する Express の方法 (呼び出されるすべてのルートに対して server.js を実行する) が原因であると考えられます。そのため、リクエストごとに の新しいインスタンスconnectionが作成され、その「準備が整った」アプリで、Express サーバーのインスタンスをもう 1 つ作成しようとします。

How to make amqp_connection accessible from any point of my app, but no doubling require("amqp").createConnection() in every point where I need to push something to RabbitMQ?

UPD: or maybe there is no problem with Express. node-amqp seems to fire ready event every second after creation. Don't know if it's correct behavior

Thank you.