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測試了一下：

2. package com.lxw1234.redis;
3.  
4. import java.util.HashMap;
5. import java.util.Map;
6. import java.util.Set;
7.  
8. import redis.clients.jedis.Jedis;
9. import redis.clients.jedis.Pipeline;
10. import redis.clients.jedis.Response;
11.  
12.  
13. public class Test {
15. public static void main(String[] args) throws Exception {
16. Jedis redis = new Jedis("127.0.0.1", 6379, 400000);
17. Map<String,String> data = new HashMap<String,String>();
18. redis.select(8);
19. redis.flushDB();
20. //hmset
21. long start = System.currentTimeMillis();
22. //直接hmset
23. for (int i=0;i<10000;i++) {
24. data.clear();
25. data.put("k_" + i, "v_" + i);
26. redis.hmset("key_" + i, data);
27. }
28. long end = System.currentTimeMillis();
29. System.out.println("dbsize:[" + redis.dbSize() + "] .. ");
30. System.out.println("hmset without pipeline used [" + (end - start) / 1000 + "] seconds ..");
31. redis.select(8);
32. redis.flushDB();
33. //使用pipeline hmset
34. Pipeline p = redis.pipelined();
35. start = System.currentTimeMillis();
36. for (int i=0;i<10000;i++) {
37. data.clear();
38. data.put("k_" + i, "v_" + i);
39. p.hmset("key_" + i, data);
40. }
41. p.sync();
42. end = System.currentTimeMillis();
43. System.out.println("dbsize:[" + redis.dbSize() + "] .. ");
44. System.out.println("hmset with pipeline used [" + (end - start) / 1000 + "] seconds ..");
46. //hmget
47. Set keys = redis.keys("*");
48. //直接使用Jedis hgetall
49. start = System.currentTimeMillis();
50. Map<String,Map<String,String>> result = new HashMap<String,Map<String,String>>();
51. for(String key : keys) {
52. result.put(key, redis.hgetAll(key));
53. }
54. end = System.currentTimeMillis();
55. System.out.println("result size:[" + result.size() + "] ..");
56. System.out.println("hgetAll without pipeline used [" + (end - start) / 1000 + "] seconds ..");
58. //使用pipeline hgetall
59. Map<String,Response<Map<String,String>>> responses = new HashMap<String,Response<Map<String,String>>>(keys.size());
60. result.clear();
61. start = System.currentTimeMillis();
62. for(String key : keys) {
63. responses.put(key, p.hgetAll(key));
64. }
65. p.sync();
66. for(String k : responses.keySet()) {
67. result.put(k, responses.get(k).get());
68. }
69. end = System.currentTimeMillis();
70. System.out.println("result size:[" + result.size() + "] ..");
71. System.out.println("hgetAll with pipeline used [" + (end - start) / 1000 + "] seconds ..");
73. redis.disconnect();
75. }
78. }
79.  

測試結果如下：

2. dbsize:[10000] ..
3. hmset without pipeline used [243] seconds ..
4. dbsize:[10000] ..
5. hmset with pipeline used [0] seconds ..
6. result size:[10000] ..
7. hgetAll without pipeline used [243] seconds ..
8. result size:[10000] ..
9. hgetAll with pipeline used [0] seconds ..

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