前言上一篇博客上已經實現了使用EventBus對具體事件行為的分發處理，某種程度上也算是基於事件驅動思想編程了。但是如上篇博客結尾處一樣，我們源碼的執行效率依然達不到心裡預期。在下單流程里我們明顯可以將部分行為進行非同步處理，提升下單操作的執行效率。 Redis基礎命令 Redis有兩種方式可支持我 ...

前言

上一篇博客上已經實現了使用EventBus對具體事件行為的分發處理，某種程度上也算是基於事件驅動思想編程了。但是如上篇博客結尾處一樣，我們源碼的執行效率依然達不到心裡預期。在下單流程里我們明顯可以將部分行為進行非同步處理，提升下單操作的執行效率。

Redis基礎命令

Redis有兩種方式可支持我們實現MQ功能，1、使用列表(List)相關命令特性；2、使用publish、subscribe命令特性；
這裡我是採取列表相關命令實現。

使用列表(List)相關命令的特性實現

壓入數據(發佈消息)
使用列表(List)的 LPUSH RPUSH 命令可以從列表左邊和右邊壓入數據;

LPUSH
將一個或多個值插入到列表頭部(此處可以將列表想象成一個從左到右的鏈表數據結構，LPUSH就是將指定的值插入最左側！)

如下命令，將多個元素壓入list1的頭部(最左側)

LPUSH list1 測試1 測試2

執行結果如下:

上面是寫入多個元素，我們也可以寫入單個元素

LPUSH list1 測試3

需要留意，每次執行完LPUSH後，Redis會返回當前列表的長度。

RPUSH
在指定列表的尾部(相當於一個鏈表的最右側)添加單個或多個元素

如下命令，還是在list1上添加多個元素，並查看執行後的list1元素信息

RPUSH list 測試4 測試5

同理，RPUSH也可直接寫入單個元素，和LPUSH一樣。

拉取數據(消費數據)
這裡的拉取數據不單單是讀取List內的元素，而是將元素從列表中取出來

BLPOP
移出並獲取列表的第一個元素(從左至右)，如果列表沒有元素會阻塞當前線程，直到等待超時或發現可彈出元素為止。

如下命令，從list1這個列表獲取從左至右第一個元素，在100秒內如果獲取則結束阻塞，否則阻塞到100秒之後。

BLPOP list1 100

執行結果如下：

需要留意的是BLPOP命令如果拉取到數據則會返回兩行數據，1行為列表的key名稱，1行為獲取到的元素值。如果直到阻塞結束都沒有獲取到元素值則直接返回命令執行超時。如下圖:

BRPOP
移出並獲取列表的最後一個元素(從左至右)，如果列表沒有元素會阻塞當前線程直到等待超時或發現可彈出元素為止。該命令與BLPOP除了獲取的元素位置不同，其他特性全部一致。

LPOP
移出並獲取列表的第一個元素(從左至右)，如獲取到元素則返回元素信息，沒有元素則立即返回null。

如下命令:

LPOP list1

RPOP
移出並獲取列表的最後一個元素(從左至右)，如獲取到元素則返回元素信息，沒有元素則立即返回null。該命令與LPOP除了獲取的元素位置不同其他特性全部一致；

RPOPLPUSH
移除列表的最後一個元素(最右側的元素)，並將該元素添加到另一個列表並返回。該命令如獲取到元素則返回元素信息，否則返回錯誤信息。

可以通過RPOPLPUSH這個命令的特性對MQ內一致性要求較高的業務進行處理，在從列表獲取元素成功後將該元素添加到一個備份列表，在業務處理完畢後再從備份列表將該元素刪除。
執行下麵命令測試下:

RPOPLPUSH list1 listback

BRPOPLPUSH
從列表中彈出一個值，將彈出的元素插入到另外一個列表中並返回它；如果列表沒有元素會阻塞列表直到等待超時或發現可彈出元素為止。

該命令其實就是在BRPOP的基礎上將LPUSH的功能加上了，依舊也保留了指定超時時間內未獲取到元素則阻塞線程。
執行下麵命令測試下:

BRPOPLPUSH list1 listback 10

執行結果如下:

完善代碼

基於上面Redis的相關命令，我們再完善下上篇博客的代碼。這裡我們需要新增一個控制台啟動項，將它作為消費服務，原來的控制台即訂單保存的控制台作為消息發佈的服務。
下單代碼更改為下麵的樣子:

        /// <summary>
        /// 非同步方式觸發訂單相關事件
        /// </summary>
        public static void AsynEventHandle()
        {
            Guid userId = Guid.NewGuid();
            Stopwatch stopwatch = new Stopwatch();
            stopwatch.Start();

            var order = new OrderModel()
            {
                CreateTime = DateTime.Now,
                Id = Guid.NewGuid(),
                Money = (decimal)300.00,
                Number = 1,
                ProductName = "鮮花一束",
                UserId = userId
            };
            Console.WriteLine($"模擬存儲訂單【採取Redis做消息隊列的非同步方式】");
            Thread.Sleep(1000);

            FullRedis fullRedis = new FullRedis("127.0.0.1:6379", "", 1);
            //這裡嘗試過使用redis 的訂閱發佈模式,在執行發佈命令時候發現值但凡出現空格或者"符號則會異常...         
            fullRedis.LPUSH("orders", new OrderModel[] { order });

            stopwatch.Stop();
            Console.WriteLine($"下單總耗時:{stopwatch.ElapsedMilliseconds}毫秒");
            Console.ReadLine();
        }

可以看到，我們已經將事件匯流排相關代碼給移除了，上面代碼除了向Redis的隊列(List)里寫入元素外就只是對訂單進行了持久化動作，所以看代碼就知道執行效率的提升了。
接下來，看消費服務的代碼。

        static void Main(string[] args)
        {
            XTrace.UseConsole();
            Console.WriteLine("進入Redis消息訂閱者模式訂單消息推送訂閱者客戶端!");

            EventBus eventBus = new EventBus();
            eventBus.EventRegister(typeof(OrderCreateEventNotifyHandle), typeof(OrderCreateEventData));
            eventBus.EventRegister(typeof(OrderCreateEventStockLockHandle), typeof(OrderCreateEventData));

            FullRedis fullRedis = new FullRedis("127.0.0.1:6379", "", 1);
            fullRedis.Log = XTrace.Log;
            fullRedis.Timeout = 30000;
            OrderModel order = null;
            while (order == null)
            { 
                order = fullRedis.BLPOP<OrderModel>("orders", 20);
                if (order != null)
                {
                    Console.WriteLine($"得到訂單信息:{JsonConvert.SerializeObject(order)}");
                    //執行相關事件
                    eventBus.Trigger(new OrderCreateEventData()
                    {
                        Order = order,
                    });
                    //再次設置為null方便迴圈讀取
                    order = null;
                }
                  
            }
            Console.ReadLine();
        }

消費服務首先從Redis里通過BLPOP從orders列表中獲取元素，再觸發事件匯流排，執行訂單保存相關業務處理。

最終看下執行效率如何？
消息發佈的執行效率(訂單保存)