Java使用Pipeline對Redis批量讀寫（hmset&hgetall）

通常狀況下，Redis Client端發出一個請求後，一般會阻塞並等待Redis服務端處理，Redis服務端處理完後請求命令後會將結果經過響應報文返回給Client。
這有點相似於HBase的Scan，一般是Client端獲取每一條記錄都是一次RPC調用服務端。
在Redis中，有沒有相似HBase Scanner Caching的東西呢，一次請求，返回多條記錄呢？
有，這就是Pipline。官方介紹 http://redis.io/topics/pipelining

經過pipeline方式當有大批量的操做時候，咱們能夠節省不少原來浪費在網絡延遲的時間，須要注意到是用pipeline方式打包命令發送，redis必須在處理完全部命令前先緩存起全部命令的處理結果。打包的命令越多，緩存消耗內存也越多。因此並非打包的命令越多越好。

使用Pipeline在對Redis批量讀寫的時候，性能上有很是大的提高。

使用Java測試了一下：

 
 
 
  
   
 
 
  
 package com.lxw1234.redis; 
 
 
  
   
 
 
  
 import java.util.HashMap; 
 
 
  
 import java.util.Map; 
 
 
  
 import java.util.Set; 
 
 
  
   
 
 
  
 import redis.clients.jedis.Jedis; 
 
 
  
 import redis.clients.jedis.Pipeline; 
 
 
  
 import redis.clients.jedis.Response; 
 
 
  
   
 
 
  
   
 
 
  
 public class Test { 
 
 
  
   
 
 
  
  public static void main(String[] args) throws Exception { 
 
 
  
  Jedis redis = new Jedis("127.0.0.1", 6379, 400000); 
 
 
  
  Map<String,String> data = new HashMap<String,String>(); 
 
 
  
  redis.select(8); 
 
 
  
  redis.flushDB(); 
 
 
  
  //hmset  
 
 
  
  long start = System.currentTimeMillis(); 
 
 
  
  //直接hmset 
 
 
  
  for (int i=0;i<10000;i++) { 
 
 
  
  data.clear(); 
 
 
  
  data.put("k_" + i, "v_" + i); 
 
 
  
  redis.hmset("key_" + i, data); 
 
 
  
  } 
 
 
  
  long end = System.currentTimeMillis(); 
 
 
  
  System.out.println("dbsize:[" + redis.dbSize() + "] .. "); 
 
 
  
  System.out.println("hmset without pipeline used [" + (end - start) / 1000 + "] seconds .."); 
 
 
  
  redis.select(8); 
 
 
  
  redis.flushDB(); 
 
 
  
  //使用pipeline hmset 
 
 
  
  Pipeline p = redis.pipelined(); 
 
 
  
  start = System.currentTimeMillis(); 
 
 
  
  for (int i=0;i<10000;i++) { 
 
 
  
  data.clear(); 
 
 
  
  data.put("k_" + i, "v_" + i); 
 
 
  
  p.hmset("key_" + i, data); 
 
 
  
  } 
 
 
  
  p.sync(); 
 
 
  
  end = System.currentTimeMillis(); 
 
 
  
  System.out.println("dbsize:[" + redis.dbSize() + "] .. "); 
 
 
  
  System.out.println("hmset with pipeline used [" + (end - start) / 1000 + "] seconds .."); 
 
 
  
   
 
 
  
  //hmget  
 
 
  
  Set keys = redis.keys("*"); 
 
 
  
  //直接使用Jedis hgetall 
 
 
  
  start = System.currentTimeMillis(); 
 
 
  
  Map<String,Map<String,String>> result = new HashMap<String,Map<String,String>>(); 
 
 
  
  for(String key : keys) { 
 
 
  
  result.put(key, redis.hgetAll(key)); 
 
 
  
  } 
 
 
  
  end = System.currentTimeMillis(); 
 
 
  
  System.out.println("result size:[" + result.size() + "] .."); 
 
 
  
  System.out.println("hgetAll without pipeline used [" + (end - start) / 1000 + "] seconds .."); 
 
 
  
   
 
 
  
  //使用pipeline hgetall 
 
 
  
  Map<String,Response<Map<String,String>>> responses = new HashMap<String,Response<Map<String,String>>>(keys.size()); 
 
 
  
  result.clear(); 
 
 
  
  start = System.currentTimeMillis(); 
 
 
  
  for(String key : keys) { 
 
 
  
  responses.put(key, p.hgetAll(key)); 
 
 
  
  } 
 
 
  
  p.sync(); 
 
 
  
  for(String k : responses.keySet()) { 
 
 
  
  result.put(k, responses.get(k).get()); 
 
 
  
  } 
 
 
  
  end = System.currentTimeMillis(); 
 
 
  
  System.out.println("result size:[" + result.size() + "] .."); 
 
 
  
  System.out.println("hgetAll with pipeline used [" + (end - start) / 1000 + "] seconds .."); 
 
 
  
   
 
 
  
  redis.disconnect(); 
 
 
  
   
 
 
  
  } 
 
 
  
   
 
 
  
   
 
 
  
 } 
 
 
  
   
 

 

測試結果以下：

 
 
 
  
   
 
 
  
 dbsize:[10000] ..  
 
 
  
 hmset without pipeline used [243] seconds .. 
 
 
  
 dbsize:[10000] ..  
 
 
  
 hmset with pipeline used [0] seconds .. 
 
 
  
 result size:[10000] .. 
 
 
  
 hgetAll without pipeline used [243] seconds .. 
 
 
  
 result size:[10000] .. 
 
 
  
 hgetAll with pipeline used [0] seconds .. 
 

 

使用pipeline來批量讀寫10000條記錄，就是小菜一碟，秒完。