redis大幅性能提升之使用管道（PipeLine）和批量（Batch）操作

       前段時間在做用戶畫像的時候，遇到了這樣的一個問題，記錄某一個商品的用戶購買群，剛好這種需求就可以用到Redis中的Set，key作為productID，value

就是具體的customerid集合，後續的話，我就可以通過productid來查看該customerid是否買了此商品，如果購買了，就可以有相關的關聯推薦，當然這隻是系統中

的一個小業務條件，這時候我就可以用到SADD操作方法，代碼如下：

        static void Main(string[] args)
        {
            ConnectionMultiplexer redis = ConnectionMultiplexer.Connect("192.168.23.151:6379");

            var db = redis.GetDatabase();

            var productID = string.Format("productID_{0}", 1);

            for (int i = 0; i < 10; i++)
            {
                var customerID = i;

                db.SetAdd(productID, customerID);
            }
        }

一：問題

    但是上面的這段代碼很明顯存在一個大問題，Redis本身就是基於tcp的一個Request/Response protocol模式，不信的話，可以用wireshark監視一下：

從圖中可以看到，有很多次的192.168.23.1 => 192.168.23.151 之間的數據往返，從傳輸內容中大概也可以看到有一個叫做productid_xxx的首碼，

那如果有百萬次區域網這樣的round trip，那這個延遲性可想而知，肯定達不到我們預想的高性能。

二：解決方案【Batch】

     剛好基於我們現有的業務，我可以定時的將批量的productid和customerid進行分組整合，然後用batch的形式插入到某一個具體的product的set中去，

接下來我可以把上面的代碼改成類似下麵這樣：

 1         static void Main(string[] args)
 2         {
 3             ConnectionMultiplexer redis = ConnectionMultiplexer.Connect("192.168.23.151:6379");
 4 
 5             var db = redis.GetDatabase();
 6 
 7             var productID = string.Format("productID_{0}", 1);
 8 
 9             var list = new List<int>();
10 
11 
12             for (int i = 0; i < 10; i++)
13             {
14                 list.Add(i);
15             }
16 
17             db.SetAdd(productID, list.Select(i => (RedisValue)i).ToArray());
18         }

從截圖中傳輸的request,response可以看到，這次我們一次性提交過去，極大的較少了在網路傳輸方面帶來的尷尬性。。

三：再次提出問題

　　product維度的畫像我們可以解決了，但是我們還有一個customerid的維度，也就是說我需要維護一個customerid為key的set集合，其中value的值為

該customerid的各種平均值，比如說“總交易次數”，“總交易金額”。。。等等這樣的聚合信息，然後推送過來的是批量的customerid，也就是說你需要定時

維護一小嘬set集合，在這種情況下某一個set的批量操作就搞不定了。。。原始代碼如下：

 1         static void Main(string[] args)
 2         {
 3             ConnectionMultiplexer redis = ConnectionMultiplexer.Connect("192.168.23.151:6379");
 4 
 5             var db = redis.GetDatabase();
 6 
 7 
 8             //批量過來的數據： customeridlist， ordertotalprice，具體業務邏輯省略
 9             var orderTotalPrice = 100;
10 
11             var customerIDList = new List<int>();
12 
13             for (int i = 0; i < 10; i++)
14             {
15                 customerIDList.Add(i);
16             }
17 
18             //foreach更新每個redis 的set集合
19             foreach (var item in customerIDList)
20             {
21                 var customerID = string.Format("customerid_{0}", item);
22 
23                 db.SetAdd(customerID, orderTotalPrice);
24             }
25         }

四：解決方案【PipeLine】

    上面這種代碼在生產上當然是行不通的，不過針對這種問題，redis早已經提出了相關的解決方案，那就是pipeline機制，原理還是一樣，將命令集整合起來通過

一條request請求一起送過去，由redis內部fake出一個client做批量執行操作，代碼如下：

 1         static void Main(string[] args)
 2         {
 3             ConnectionMultiplexer redis = ConnectionMultiplexer.Connect("192.168.23.151:6379");
 4 
 5             var db = redis.GetDatabase();
 6 
 7 
 8             //批量過來的數據： customeridlist， ordertotalprice，具體業務邏輯省略
 9             var orderTotalPrice = 100;
10 
11             var customerIDList = new List<int>();
12 
13             for (int i = 0; i < 10; i++)
14             {
15                 customerIDList.Add(i);
16             }
17 
18             var batch = db.CreateBatch();
19 
20             foreach (var item in customerIDList)
21             {
22                 var customerID = string.Format("customerid_{0}", item);
23 
24                 batch.SetAddAsync(customerID, orderTotalPrice);
25             }
26 
27             batch.Execute();
28         }

然後，我們再看下麵的wireshark截圖，可以看到有很多的SADD這樣的小命令，這就說明有很多命令是一起過去的，大大的提升了性能。

最後可以再看一下redis，數據也是有的，是不是很爽~~~

192.168.23.151:6379> keys *
 1) "customerid_0"
 2) "customerid_9"
 3) "customerid_1"
 4) "customerid_3"
 5) "customerid_8"
 6) "customerid_2"
 7) "customerid_7"
 8) "customerid_5"
 9) "customerid_6"
10) "customerid_4"

好了，先就說到這裡了，希望本篇對你有幫助。