Redis使用pipeLine批量获取数据加快接口响应速度

一、背景

需求：

redis通过tcp来对外提供服务，client通过socket连接发起请求，每个请求在命令发出后会阻塞等待redis服务器进行处理，处理完毕后将结果返回给client。

其实和一个http的服务器类似，一问一答，请求一次给一次响应。而这个过程在排除掉redis服务本身做复杂操作时的耗时的话，可以看到最耗时的就是这个网络传输过程。每一个命令都对应了发送、接收两个网络传输，假如一个流程需要0.1秒，那么一秒最多只能处理10个请求，将严重制约redis的性能。

在很多场景下，我们要完成一个业务，可能会对redis做连续的多个操作，譬如库存减一、订单加一、余额扣减等等，这有很多个步骤是需要依次连续执行的。
潜在隐患：这样的场景，网络传输的耗时将是限制redis处理量的主要瓶颈。循环key，获取value，可能会造成连接池的连接数增多，连接的创建和摧毁，消耗性能
解决方法：

可以引入pipeline了，pipeline管道就是解决执行大量命令时、会产生大量同学次数而导致延迟的技术。

其实原理很简单，pipeline就是把所有的命令一次发过去，避免频繁的发送、接收带来的网络开销，redis在打包接收到一堆命令后，依次执行，然后把结果再打包返回给客户端。

根据项目中的缓存数据结构的实际情况，数据结构为string类型的，使用RedisTemplate的multiGet方法；数据结构为hash，使用Pipeline(管道)，组合命令，批量操作redis。

二、操作

RedisTemplate的multiGet的操作
- 针对数据结构为String类型
- 示例代码
List keys = new ArrayList<>(); for (Book e : booklist) { String key = generateKey.getKey(e); keys.add(key); } List resultStr = template.opsForValue().multiGet(

2.RedisTemplate的Pipeline使用

为什么Pipelining这么快？

先看看原来的多条命令，是如何执行的：

Redis Client->>Redis Server: 发送第1个命令

Redis Server->>Redis Client: 响应第1个命令

Redis Client->>Redis Server: 发送第2个命令

Redis Server->>Redis Client: 响应第2个命令

Redis Client->>Redis Server: 发送第n个命令

Redis Server->>Redis Client: 响应第n个命令

Pipeling机制是怎样的呢：
    Redis Client->>Redis Server: 发送第1个命令（缓存在Redis Client，未即时发送）
    Redis Client->>Redis Server: 发送第2个命令（缓存在Redis Client，未即时发送）
    Redis Client->>Redis Server: 发送第n个命令（缓存在Redis Client，未即时发送）
    Redis Client->>Redis Server: 发送累积的命令
    Redis Server->>Redis Client: 响应第1、2、n个命令

示例代码

package cn.chinotan.controller;

import cn.chinotan.service.RedisService; import lombok.extern.java.Log; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit;

/**

@program: test
@description: redis批量数据测试
@author: xingcheng
@create: 2019-03-16 16:26

**/ @RestController @RequestMapping("/redisBatch") @Log public class RedisBatchController {

@Autowired
StringRedisTemplate redisTemplate;

@Autowired
Map<String, RedisService> redisServiceMap;

/**
 * VALUE缓存时间 3分钟
 */
public static final Integer VALUE_TIME = 1;

/**
 * 测试列表长度
 */
public static final Integer SIZE = 100000;

@GetMapping(value = "/test/{model}")
public Object hello(@PathVariable("model") String model) {
    List<Map<String, String>> saveList = new ArrayList<>(SIZE);
    List<String> keyList = new ArrayList<>(SIZE);
    for (int i = 0; i < SIZE; i++) {
        Map<String, String> objectObjectMap = new HashMap<>();
        String key = String.valueOf(i);
        objectObjectMap.put("key", key);
        StringBuilder sb = new StringBuilder();
        objectObjectMap.put("value", sb.append("value").append(i).toString());
        saveList.add(objectObjectMap);
        // 记录全部key
        keyList.add(key);
    }
    
    // 获取对应的实现
    RedisService redisService = redisServiceMap.get(model);
    
    long saveStart = System.currentTimeMillis();
    redisService.batchInsert(saveList, TimeUnit.MINUTES, VALUE_TIME);
    long saveEnd = System.currentTimeMillis();
    log.info("插入耗时：" + (saveEnd - saveStart) + " ms");
    // 批量获取
    long getStart = System.currentTimeMillis();
    List<String> valueList = redisService.batchGet(keyList);
    long getEnd = System.currentTimeMillis();
    log.info("获取耗时：" + (getEnd - getStart) + " ms");
    return valueList;
}

}

package cn.chinotan.controller;

import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit;

/**

@program: test
@description: redis批量数据测试
@author: xingcheng
@create: 2019-03-16 16:26

**/ @RestController @RequestMapping("/redisBatch") @Log public class RedisBatchController {

@Autowired
StringRedisTemplate redisTemplate;

@Autowired
Map<String, RedisService> redisServiceMap;

/**
 * VALUE缓存时间 3分钟
 */
public static final Integer VALUE_TIME = 1;

/**
 * 测试列表长度
 */
public static final Integer SIZE = 100000;

@GetMapping(value = "/test/{model}")
public Object hello(@PathVariable("model") String model) {
    List<Map<String, String>> saveList = new ArrayList<>(SIZE);
    List<String> keyList = new ArrayList<>(SIZE);
    for (int i = 0; i < SIZE; i++) {
        Map<String, String> objectObjectMap = new HashMap<>();
        String key = String.valueOf(i);
        objectObjectMap.put("key", key);
        StringBuilder sb = new StringBuilder();
        objectObjectMap.put("value", sb.append("value").append(i).toString());
        saveList.add(objectObjectMap);
        // 记录全部key
        keyList.add(key);
    }
    
    // 获取对应的实现
    RedisService redisService = redisServiceMap.get(model);
    
    long saveStart = System.currentTimeMillis();
    redisService.batchInsert(saveList, TimeUnit.MINUTES, VALUE_TIME);
    long saveEnd = System.currentTimeMillis();
    log.info("插入耗时：" + (saveEnd - saveStart) + " ms");
    // 批量获取
    long getStart = System.currentTimeMillis();
    List<String> valueList = redisService.batchGet(keyList);
    long getEnd = System.currentTimeMillis();
    log.info("获取耗时：" + (getEnd - getStart) + " ms");
    return valueList;
}

}

package cn.chinotan.service;

import org.springframework.beans.factory.annotation.Autowired; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.StringRedisConnection; import org.springframework.data.redis.core.RedisCallback; import org.springframework.data.redis.core.RedisOperations; import org.springframework.data.redis.core.SessionCallback; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Service;

import java.util.*; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors;

/**

@program: test
@description: redis管道操作
@author: xingcheng
@create: 2019-03-16 16:47

**/ @Service("pipe") public class RedisPipelineService implements RedisService {

@Autowired
StringRedisTemplate redisTemplate;

@Override
public void batchInsert(List<Map<String, String>> saveList, TimeUnit unit, int timeout) {
    /* 插入多条数据 */
    redisTemplate.executePipelined(new SessionCallback<Object>() {
        @Override
        public <K, V> Object execute(RedisOperations<K, V> redisOperations) throws DataAccessException {
            for (Map<String, String> needSave : saveList) {
                redisTemplate.opsForValue().set(needSave.get("key"), needSave.get("value"), timeout,unit);
            }
            return null;
        }
    });
}

@Override
public List<String> batchGet(List<String> keyList) {
    /* 批量获取多条数据 */
    List<Object> objects = redisTemplate.executePipelined(new RedisCallback<String>() {
        @Override
        public String doInRedis(RedisConnection redisConnection) throws DataAccessException {
            StringRedisConnection stringRedisConnection = (StringRedisConnection) redisConnection;
            for (String key : keyList) {
                stringRedisConnection.get(key);
            }
            return null;
        }
    });

    List<String> collect = objects.stream().map(val -> String.valueOf(val)).collect(Collectors.toList());

    return collect;
}

}

package cn.chinotan.service;

import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Service;

import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit;

/**

@program: test
@description: redis普通遍历操作
@author: xingcheng
@create: 2019-03-16 16:47

**/ @Service("generic") public class RedisGenericService implements RedisService {

@Autowired
StringRedisTemplate redisTemplate;

@Override
public void batchInsert(List<Map<String, String>> saveList, TimeUnit unit, int timeout) {
    for (Map<String, String> needSave : saveList) {
        redisTemplate.opsForValue().set(needSave.get("key"), needSave.get("value"), timeout,unit);
    }
}

@Override
public List<String> batchGet(List<String> keyList) {
    List<String> values = new ArrayList<>(keyList.size());
    for (String key : keyList) {
        String value = redisTemplate.opsForValue().get(key);
        values.add(value);
    }
    return values;
}

}

测试结果：

Redis使用pipeLine批量获取数据加快接口响应速度

可以看到性能提升了20倍之多

基于其特性，它有两个明显的局限性：

鉴于Pipepining发送命令的特性，Redis服务器是以队列来存储准备执行的命令，而队列是存放在有限的内存中的，所以不宜一次性发送过多的命令。如果需要大量的命令，可分批进行，效率不会相差太远滴，总好过内存溢出嘛~~
由于pipeline的原理是收集需执行的命令，到最后才一次性执行。所以无法在中途立即查得数据的结果（需待pipelining完毕后才能查得结果），这样会使得无法立即查得数据进行条件判断（比如判断是非继续插入记录）。

Android进阶之旅-(NDK实战篇之数据结构算法进阶)

Andriod第三方源码分析

热门文章