.NET客戶端實現Redis中的管道（PipeLine）與事物（Transactions）

時間 2019-11-06

標籤客戶端實現 redis 管道 pipeline 事物 transactions 欄目 Redis 简体版

原文原文鏈接

序言

Redis中的管道（PipeLine）特性：簡述一下就是，Redis如何從客戶端一次發送多個命令，服務端到客戶端如何一次性響應多個命令。git

Redis使用的是客戶端-服務器模型和請求/響應協議的TCP服務器,這就意味着一個請求要有如下步驟才能完成：一、客戶端向服務器發送查詢命令，而後一般以阻塞的方式等待服務器相應。二、服務器處理查詢命令，並將相應發送回客戶端。這樣便會經過網絡鏈接，若是是本地迴環接口那麼就能特別迅速的響應，可是若是走外網，甚至外網再作一系列的層層轉發，那就顯的格外蛋疼。不管網絡延時是多少，那麼都將佔用總體響應的時間。這樣一來若是一次發送1個命令，網絡延時爲100ms，咱們不得不作。那麼若是1次發1000個命令，那麼網絡延時100*1000ms就很難容忍啦。github

針對與上面的問題，Redis在2.6版本之後就都提供啦管道（Pipeline）功能。他可使客戶端在沒有讀取舊的響應時，處理新的請求。這樣即可以向服務器發送多個命令，而沒必要等待答覆，直到最後一個步驟中讀取答覆。這被稱爲管線（PipeLine），而且是幾十年來普遍使用的技術。例如，許多POP3協議實現已經支持此功能，大大加快了從服務器下載新電子郵件的過程。redis

那麼事務這個詞彙，常常遇到，就很少唧唧啦，目標要一致就好，即是一組操做怎麼作成原子性操做，使他去不了終點，回到原點。數據庫

簡述wireshark抓包工具

爲啦讓你們對管線有更形象的感觀，這一節咱們先說說Wireshark抓包工具，他會讓你看到客戶端到服務器經過tcp協議發送的redis命令的過程與詳細。服務器

wireshark可以撲捉到系統發送和接受的每個報文，咱們這裏只作一些過濾的簡述。下圖就是他的樣子，你打開後能夠能夠摸索下他的用法。網絡

簡述幾個過濾規則：異步

一、ip過濾：目標ip過濾：ip.dst==172.18.8.11,源ip地址過濾：ip.src==192.168.1.12;async

二、端口過濾：tcp.port==80，這條規則是把源端口和目的端口爲80的都過濾出來。使用tcp.dstport==80只過濾目的端口爲80的，tcp.srcport==80只過濾源端口爲80的包；tcp

三、協議過濾：直接在fiter框中輸入協議名稱便可，如：http，tcp，udp，...工具

四、http模式過濾:過濾get包，http.request.method=="GET",過濾post包，http.request.method=="POST"；

五、若是使用多條件過濾，則須要加鏈接符號，and。好比 ip.src==192.168.1.12 and http.request.method=="POST" and tcp.srcport==80

StackExchange.Redis實現Redis管線（Pipeline）

上兩張圖片管線便一目瞭然啦。

客戶端對redis服務器進行屢次請求的話，通常普通模式是這樣子的

客戶端對redis服務器進行屢次請求的話，管道模式是這樣子的

通常模式咱們上代碼：

 public static void GetNoPipelining()
        {
            for (var i = 0; i < 3; i++)
            {
                var key = "name:" + i;
                db.StringAppend(key, "張龍豪");
            }
        }

查看tcp請求報文的data

這樣你本身作的過程當中，能夠看到我圈起來的3個tcp請求的key分別爲name：0，name:1,name:2這樣子。

那麼咱們使用管道模式

 public static void GetPipelining()
        {
            var batch = db.CreateBatch();
            for (int i = 0; i < 3; i++)
            {
                var key = "mename:" + i;
                batch.StringAppendAsync(key, "張龍豪");
            }
            batch.Execute();
        }

再來看下請求

這樣很明顯就能看出來是1個請求發送出來啦多個命令。那麼咱們不用createBatch（）也是能夠實現這樣的效果的。

            var a = db.StringAppendAsync("zlh:1", "zhanglonghao1");
            var b = db.StringAppendAsync("zlh:2", "zhanglonghao2");
            var c = db.StringAppendAsync("zlh:3", "zhanglonghao3");
            var aa = db.Wait(a);
            var bb = db.Wait(a);
            var cc = db.Wait(a);

在接下來咱們作一個簡單的性能比較。代碼以下：

  static void Main(string[] args)
        {
            Stopwatch watch = new Stopwatch();
            Stopwatch watch1 = new Stopwatch();
            watch.Start();
            GetNoPipelining();
            Console.WriteLine("通常循環耗時：" + watch.ElapsedMilliseconds);
            watch.Stop();
            watch1.Start();
            GetPipelining();
            Console.WriteLine("Pipelining插入耗時：" + watch1.ElapsedMilliseconds);
            watch1.Stop();
            Console.ReadLine();
        }

        public static void GetNoPipelining()
        {
            for (var i = 0; i < 5000; i++)
            {
                var key = "name:" + i;
                db.StringAppend(key, "張龍豪");
            }
        }
        public static void GetPipelining()
        {
            var batch = db.CreateBatch();
            for (int i = 0; i < 5000; i++)
            {
                var key = "mename:" + i;
                batch.StringAppendAsync(key, "張龍豪");
            }
            batch.Execute();
        }

結果以下：

到此我還要說一下StackExchange.Redis的三種命令模式，其中使用2和3的模式發送命令，會默認被封裝在管道中，不信的話，你能夠作個小demo測試下：

一、sync：同步模式，會直接阻塞調用者，但不會阻塞其餘線程。

二、async：異步模式，使用task模型封裝。

三、fire-and-forget：發送命令，而後徹底不關心最終何時完成命令操做。在Fire-and-Forget模式下,全部命令都會當即獲得返回值，該值都是該返回值類型的默認值,好比操做返回類型是bool將會當即獲得false,由於false = default(bool)。

此節參考redis官方文檔與StackExchange.Redis官方文檔，鏈接以下：

https://redis.io/topics/pipelining

https://github.com/StackExchange/StackExchange.Redis/blob/master/Docs/PipelinesMultiplexers.md

StackExchange.Redis實現Redis事務（Transactions）

這個看官方文檔，我只能說實現的很奇怪吧。我先描述下個人環境，就是準備一個空redis庫，而後一步一步往下走，咱們寫代碼看結果，來搞一搞這個事務。

 static void Main(string[] args)
        {
            var tran = db.CreateTransaction();
            tran.AddCondition(Condition.ListIndexNotEqual("zlh:1",0,"zhanglonghao"));        
            tran.ListRightPushAsync("zlh:1", "zhanglonghao"); 
            bool committed = tran.Execute();
            Console.WriteLine(committed);
            Console.ReadLine();
        }

執行結果爲：true。數據庫中結果以下，說明咱們插入成功。

即：若是key爲：zlh：1的list集合在索引0初的value!=zhanglonghao的話，咱們從鏈表右側插入一條數據key爲zlh：1value爲zhanglonghao，成功。由於第一次操做爲空庫。0處確實不爲張龍豪。

數據不清空，繼續上代碼。

 static void Main(string[] args)
        {
            var tran = db.CreateTransaction();
            tran.AddCondition(Condition.ListIndexNotEqual("zlh:1",0,"zhanglonghao"));        
            tran.ListRightPushAsync("zlh:1", "zhanglonghao1");           
            bool committed = tran.Execute();
            Console.WriteLine(committed);
            Console.ReadLine();
        }

結果爲false，數據庫沒有增減數據。已久與上圖的數據保持一致。

緣由分析：0處此時爲zhanglonghao，因此ListIndexNotEqual("zlh:1",0,"zhanglonghao")爲假命題，直接回滾，不執行下面的插入命令。

數據不清空，繼續上代碼：

 static void Main(string[] args)
        {
            var tran = db.CreateTransaction();
            tran.AddCondition(Condition.ListIndexEqual("zlh:1",0,"zhanglonghao"));        
            tran.ListRightPushAsync("zlh:1", "zhanglonghao1");           
            bool committed = tran.Execute();
            Console.WriteLine(committed);
            Console.ReadLine();
        }

結果爲true，數據結果以下，增加一條值爲zhanglonghao1的數據：

緣由分析：ListIndexEqual("zlh:1",0,"zhanglonghao")爲真命題，執行下面的操做，提交事物。

數據不刪繼續上代碼：

static void Main(string[] args)
        {
            var tran = db.CreateTransaction();
            tran.AddCondition(Condition.ListIndexEqual("zlh:1",0,"zhanglonghao"));        
            tran.ListRightPushAsync("zlh:1", "zhanglonghao2");
            tran.AddCondition(Condition.ListIndexNotEqual("zlh:1", 0, "zhanglonghao"));
            tran.ListRightPushAsync("zlh:1", "zhanglonghao3");
            bool committed = tran.Execute();
            Console.WriteLine(committed);
            Console.ReadLine();
        }