详解Java使用Pipeline对Redis批量读写（hmset&hgetall）

一般情况下，redis client端发出一个请求后，通常会阻塞并等待redis服务端处理，redis服务端处理完后请求命令后会将结果通过响应报文返回给client。

感觉这有点类似于hbase的scan，通常是client端获取每一条记录都是一次rpc调用服务端。

在redis中，有没有类似hbase scanner caching的东西呢，一次请求，返回多条记录呢？

有，这就是pipline。官方介绍

通过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<string> 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条记录，就是小菜一碟，秒完。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。