Redis學習筆記~分布式的Pub/Sub模式 - lori－IT工程師數位筆記本

文章出處

redis的客戶端有很多，這次用它的pub/sub發布與訂閱我選擇了StackExchange.Redis，發布與訂閱大家應該很清楚了，首先一個訂閱者，訂閱一個服務，服務執行一些處理程序（可能是寫個日志，插入個數據，發個email）然后當另一個項目的某個業務發布這個服務后，被訂閱的程序將會被執行，這個聽起來很有意思，redis有好的實現了這個pub/sub功能。

看一下結構圖

Sub訂閱（消息消費者）

對于訂閱方，這里類似于一個服務，它會長期運行著，被啟動后動態訂閱一些服務進來，以便以后再被其它服務調用

   　　　　　//sub a function in A-project
            PubSubManager.Instance.Subscribe("UserLog", (msg) =>
            {
                //訂閱者處理自己的業務邏輯，這相關于隊列服務要干的事
                Console.WriteLine(msg);
            });

Pub發布（消息生產者）

而對于其它項目，如Ａ網站，它可能在被在ＰＯＳＴ動作后發布這個UserLog的服務，這時上面的訂閱方將會消費它（消費者模式），或者服務代碼被執行！

 　　　　[HttpPost]
        public ActionResult Index(string user)
        {
            PubSubManager.Instance.Publish("UserLog", user + "這個用戶提交表單了");
            return View();
        }

對于一直運行的服務，將會收到來自不同項目的消息，而它只負責消費它！

Lind.DDD.PublishSubscribe

封裝了一些標準的pub/sub方法，它可以有多種實現方法，本例使用redis這個中間件

　　 /// <summary>
    /// 發布訂閱的接口規則
    /// </summary>
    public interface IPubSub
    {
        /// <summary>
        /// 發布，有順序，對象源是字符串
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="value"></param>
        void Publish(string channel, string value);
        /// <summary>
        /// 訂閱，對象源是字符串
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="action"></param>
        void Subscribe(string channel, Action<string> action);
        /// <summary>
        /// 異步發布，無順序，對象源是字符串
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="value"></param>
        void PublishAsync(string channel, string value);
        /// <summary>
        /// 異步訂閱，無順序，對象源是字符串
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="action"></param>
        void SubscribeAsync(string channel, Action<string> action);

        /// <summary>
        /// 發布，有順序，對象源是Byte[]
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="value"></param>
        void PublishByte(string channel, byte[] value);
        /// <summary>
        /// 訂閱，對象源是Byte[]
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="action"></param>
        void SubscribeByte(string channel, Action<byte[]> action);
        /// <summary>
        /// 異步發布，有順序，對象源是Byte[]
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="value"></param>
        void PublishByteAsync(string channel, byte[] value);
        /// <summary>
        /// 異步訂閱，對象源是Byte[]
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="action"></param>
        void SubscribeByteAsync(string channel, Action<byte[]> action);

        /// <summary>
        /// 發布，有順序，對象源是泛型對象
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="channel"></param>
        /// <param name="value"></param>
        void Publish<T>(string channel, T value);
        /// <summary>
        /// 訂閱，對象源是泛型對象
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="channel"></param>
        /// <param name="action"></param>
        void Subscribe<T>(string channel, Action<T> action);
        /// <summary>
        /// 異步發布，有順序，對象源是泛型對象
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="channel"></param>
        /// <param name="value"></param>
        void PublishAsync<T>(string channel, T value);
        /// <summary>
        /// 異步訂閱，對象源是泛型對象
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="channel"></param>
        /// <param name="action"></param>
        void SubscribeAsync<T>(string channel, Action<T> action);
        /// <summary>
        /// 取消指定訂閱
        /// </summary>
        /// <param name="channel"></param>
        void UnSubscribe(string channel);
        /// <summary>
        /// 取消所有訂閱
        /// </summary>
        /// <param name="channel"></param>
        void UnSubscribeAll();
    }