C# は非同期メッセージ キューを実装します

メッセージキュー

メッセージキュー（英語: Message queue ）は、プロセス間通信、または同じプロセスの異なるスレッド間の通信の方法であり、通常はユーザーからの一連の入力を処理するために使用されます。メッセージ キューは、非同期通信プロトコルを提供します。キュー内の各レコードには、発生時刻、入力デバイスの種類、特定の入力パラメータなどの詳細情報が含まれます。つまり、メッセージの送信者と受信者は必要ありません。同時にメッセージキューと対話します。メッセージは、受信者が取得するまでキュー内に残ります。

簡単に言うと、キューは処理する必要があるコマンドを格納しますが、処理結果は時間内に取得されません

実際、メッセージ キューはリンク リスト構造に格納されることがよくあります。権限を持つプロセスは、メッセージ キューにメッセージを書き込んだり、メッセージ キューからメッセージを読み取ったりできます。

現在、JBoss Messaging、JORAM、Apache ActiveMQ、Sun Open Message Queue、Apache Qpid、HTTPSQS など、メッセージ キューのオープンソース実装が多数あります。

長所、短所

メッセージ キュー自体は非同期であるため、受信者は送信されてからかなり時間が経ってもメッセージを取得できますが、これはほとんどの通信プロトコルとは異なります。たとえば、WWW で使用される HTTP プロトコルは、クライアントがリクエストを行った後にサーバーが応答するまで待機する必要があるため、同期プロトコルです。ただし、非同期通信プロトコルが必要になる状況は数多くあります。たとえば、あるプロセスは別のプロセスにイベントが発生したことを通知しますが、応答を待つ必要はありません。ただし、メッセージ キューの非同期の性質により、受信者が最新のメッセージを受信するためにメッセージ キューをポーリングする必要があるという欠点も生じます。

シグナルと比較して、メッセージキューはより多くの情報を送信できます。パイプと比較して、メッセージ キューはフォーマットされたデータを提供するため、開発者の作業負荷を軽減できます。ただし、メッセージ キューには依然としてサイズ制限があります。

キューメッセージの読み取り

には主に 2 つのタイプがあります (1) サーバープッシュ、(2) クライアントプル: 主にクライアントが定期的にポーリングしてメッセージ処理を取り除きます。処理のためのサブスクライバー

メッセージのストレージ

は、Redis などの DB を使用することもできます。

非同期処理の一貫性を確保する方法

キューの主な目的はメッセージを保存することであり、呼び出しと実装を非同期にすることもできます。ただし、メッセージ処理の一貫性を確保したい場合は、マスターとスレーブの DB を操作し、マスターが書き込みを担当し、スレーブが読み取りを担当するなど、業務処理の順序を区別するのが良い方法です。同時に、複数の中間状態が呼び出し結果に一致する場合、それらはビジネス時に処理されます。 、「例外メッセージ」は次の操作まで保持されます。

コードをアップロードします

メッセージ反対コアキューを確立します

{    public delegate void MessageQueueEventNotifyHandler(Message.BaseMessage message);
 
    public class MessageQueue:Queue<BaseMessage>
    {
        public static MessageQueue GlobalQueue = new MessageQueue();
 
        private Timer timer = new Timer();
        public MessageQueue() {
            this.timer.Interval = 5000;
            this.timer.Elapsed += Notify;
            this.timer.Enabled = true;
        }
        private void Notify(object sender, ElapsedEventArgs e) {
            lock (this) {
                if (this.Count > 0) {
                    //this.messageNotifyEvent.GetInvocationList()[0].DynamicInvoke(this.Dequeue());
                    var message = this.Dequeue();
                    this.messageNotifyEvent(message);
                }
            }
        }
 
        private MessageQueueEventNotifyHandler messageNotifyEvent;
        public event MessageQueueEventNotifyHandler MessageNotifyEvent {
            add {
                this.messageNotifyEvent += value;
            }
 
            remove {
                if (this.messageNotifyEvent != null) {
                    this.messageNotifyEvent -= value;
                }
            }
        }
    }
}

イベント処理

public const string OrderCodePrefix = "P";        public void Submit(Message.BaseMessage message)
        {
            Order order = message.Body as Order;
 
            if (order.OrderCode.StartsWith(OrderCodePrefix))
            {
                System.Console.WriteLine("这个是个正确的以({0})开头的订单:{1}", OrderCodePrefix,order.OrderCode);
            }
            else {
                System.Console.WriteLine("这个是个错误的订单，没有以({0})开头:{1}",OrderCodePrefix,order.OrderCode);
            }
        }

は特定のビジネスに応じてカスタマイズできます;

追加Proxy

public class OrderServiceProxy:IOrderService    {
        public void Submit(Message.BaseMessage message)
        {
            MessageQueue.MessageQueue.GlobalQueue.Enqueue(message);
        }
    }

Client call

OrderService orderService = new OrderService();            MessageQueue.MessageQueue.GlobalQueue.MessageNotifyEvent += orderService.Submit;
 
            var orders = new List<Order>() {
                new Order(){OrderCode="P001"},
                new Order(){OrderCode="P002"},
                new Order(){OrderCode="B003"}
            };
 
            OrderServiceProxy proxy = new OrderServiceProxy();
            orders.ForEach(order => proxy.Submit(new Message.BaseMessage() { Body=order}));
 
            Console.ReadLine();

を介してメッセージをキューに送信します。これは、イベントバインディングとパーソナライズされた処理のトリガーのニーズを満たし、同時にメッセージの非同期化の目的を達成しました。これをさらに詳細に拡張して、その後のプロジェクト。

上記は、非同期メッセージキューを実装するための C# の内容です。さらに関連する内容については、PHP 中国語 Web サイト (www.php.cn) に注目してください。