Java Redis的Pipeline管道，批量操作，节省大量网络往返时间 & Redis批量读写（hmset&hgetall） 使用Pipeline

时间：2023-12-27 14:39:01

一般情况下，大家使用redis去put/get都是先拿到一个jedis实例，然后操作，然后释放连接；这种模式是

请求-响应，请求-响应

这种模式，下一次请求必须得等第一次请求响应回来之后才可以，因为redis是单线程的，按部就班，一步一步来。

而pipeline管道改变了这种请求模式，客户端可以一次发送多个命令，无须等待服务器的返回，

请求，请求，请求，响应，响应，响应

这种模式

这就大大减少了影响性能的关键因素-网络往返时间

下面就上面两种模式以及JDK的map三者做一个性能比较

  1. package redis;
  3. import java.util.concurrent.BlockingQueue;
  4. import java.util.concurrent.LinkedBlockingQueue;
  6. import redis.clients.jedis.ShardedJedis;
  7. import redis.clients.jedis.ShardedJedisPipeline;
  9. /**
  10. * @Type ShardRedisDemo.java
  11. * @Desc
  12. * @author chiwei
  13. * @date 2016年6月13日 下午3:24:25
  14. * @version
  15. */
  16. public class ShardRedisDemo {
  18. public static void main(String[] args) throws InterruptedException {
  19. ShardRedisClient src = new ShardRedisClient();
  20. src.setServers("redis://172.23.26.135:7379");
  21. src.init();
  22. int count = 10000;
  23. ShardedJedis sj = src.getResource();
  24. long begin = System.currentTimeMillis();
  25. for (int i = 0; i < count; i++) {
  26. sj.set("a" + i, "v" + i);
  27. }
  28. sj.close();
  29. System.out.println(System.currentTimeMillis() - begin);
  30. sj = src.getResource();
  31. ShardedJedisPipeline p = sj.pipelined();
  32. begin = System.currentTimeMillis();
  33. for (int i = 0; i < count; i++) {
  34. p.set("ap" + i, "vp" + i);
  35. }
  36. p.sync();
  37. sj.close();
  38. System.out.println(System.currentTimeMillis() - begin);
  39. BlockingQueue<String> logQueue = new LinkedBlockingQueue<String>();
  40. begin = System.currentTimeMillis();
  41. for (int i = 0; i < count; i++) {
  42. logQueue.put("i=" + i);
  43. }
  44. System.out.println(System.currentTimeMillis() - begin);
  45. }
  47. }
  49. /**
  50. * Revision history
  51. * -------------------------------------------------------------------------
  52. *
  53. * Date Author Note
  54. * -------------------------------------------------------------------------
  55. * 2016年6月13日 chiwei create
  56. */

结果如下：

45027

116

11

大家看相对时间就行了，我测试时是经过VPN连的redis，由此结果可见pipeline的性能惊人的高。

但是pipeline适合于什么样的场景使用呢？

有些系统可能对可靠性要求很高，每次操作都需要立马知道这次操作是否成功，是否数据已经写进redis了，那这种场景就不适合。

还有的系统，可能是批量的将数据写入redis，允许一定比例的写入失败，那么这种场景就可以使用了，比如10000条一下进入redis，可能失败了2条无所谓，后期有补偿机制就行了，比如短信群发这种场景，如果一下群发10000条，按照第一种模式去实现，那这个请求过来，要很久才能给客户端响应，这个延迟就太长了，如果客户端请求设置了超时时间5秒，那肯定就抛出异常了，而且本身群发短信要求实时性也没那么高，这时候用pipeline最好了。

=======================================================================================================================

一般情况下，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测试了一下：

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

测试结果如下：

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

使用pipeline来批量读写10000条记录，就是小菜一碟，秒完。

转自：https://blog.csdn.net/simonchi/article/details/52231674

转自：https://www.cnblogs.com/an7ing/p/5082243.html

相关文章