redis 批量操作
一、背景
- 需求:
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<String> keys = new ArrayList<>(); for (Book e : booklist) { String key = generateKey.getKey(e); keys.add(key); } List<Serializable> resultStr = redisTemplate.opsForValue().multiGet(keys)
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.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; } }
測試結果:
可以看到效能提升了20倍之多
基於其特性,它有兩個明顯的侷限性:
- 鑑於Pipepining傳送命令的特性,Redis伺服器是以佇列來儲存準備執行的命令,而佇列是存放在有限的記憶體中的,所以不宜一次性發送過多的命令。如果需要大量的命令,可分批進行,效率不會相差太遠滴,總好過記憶體溢位嘛~~
- 由於pipeline的原理是收集需執行的命令,到最後才一次性執行。所以無法在中途立即查得資料的結果(需待pipelining完畢後才能查得結果),這樣會使得無法立即查得資料進行條件判斷(比如判斷是非繼續插入記錄)。