Redis Pipelining 底層原理分析及實踐
Redis是一種基于客戶端-服務(wù)端模型以及請求/響應(yīng)的TCP服務(wù)。在遇到批處理命令執(zhí)行時，Redis提供了Pipelining(管道)來提升批處理性能。本文結(jié)合實踐分析了Spring Boot框架下Redis的Lettuce客戶端和Redisson客戶端對Pipeline特性的支持原理，并針對實踐過程中遇到的問題進行了分析，可以幫助開發(fā)者了解不同客戶端對Pipeline支持原理及避免實際使用中出現(xiàn)問題。
一、前言
Redis 已經(jīng)提供了像 mget 、mset 這種批量的命令，但是某些操作根本就不支持或沒有批量的操作，從而與 Redis 高性能背道而馳。為此, Redis基于管道機制，提供Redis Pipeline新特性。Redis Pipeline是一種通過一次性發(fā)送多條命令并在執(zhí)行完后一次性將結(jié)果返回，從而減少客戶端與redis的通信次數(shù)來實現(xiàn)降低往返延時時間提升操作性能的技術(shù)。目前，Redis Pipeline是被很多個版本的Redis 客戶端所支持的。
二、Pipeline 底層原理分析
2.1 Redis單個命令執(zhí)行基本步驟
Redis是一種基于客戶端-服務(wù)端模型以及請求/響應(yīng)的TCP服務(wù)。一次Redis客戶端發(fā)起的請求，經(jīng)過服務(wù)端的響應(yīng)后，大致會經(jīng)歷如下的步驟：
- 客戶端發(fā)起一個（查詢/插入）請求，并監(jiān)聽socket返回，通常情況都是阻塞模式等待Redis服務(wù)器的響應(yīng)。
- 服務(wù)端處理命令，并且返回處理結(jié)果給客戶端。
- 客戶端接收到服務(wù)的返回結(jié)果，程序從阻塞代碼處返回。
2.2 RTT 時間
Redis客戶端和服務(wù)端之間通過網(wǎng)絡(luò)連接進行數(shù)據(jù)傳輸，數(shù)據(jù)包從客戶端到達服務(wù)器，并從服務(wù)器返回數(shù)據(jù)回復客戶端的時間被稱之為RTT(Round Trip Time - 往返時間)。我們可以很容易就意識到，Redis在連續(xù)請求服務(wù)端時，如果RTT時間為250ms, 即使Redis每秒能處理100k請求，但也會因為網(wǎng)絡(luò)傳輸花費大量時間，導致每秒最多也只能處理4個請求，導致整體性能的下降。
2.3 Redis Pipeline
為了提升效率，這時候Pipeline出現(xiàn)了。Pipelining不僅僅能夠降低RRT，實際上它極大的提升了單次執(zhí)行的操作數(shù)。這是因為如果不使用Pipelining，那么每次執(zhí)行單個命令,從訪問數(shù)據(jù)的結(jié)構(gòu)和服務(wù)端產(chǎn)生應(yīng)答的角度，它的成本是很低的。但是從執(zhí)行網(wǎng)絡(luò)IO的角度，它的成本其實是很高的。其中涉及到read()和write()的系統(tǒng)調(diào)用，這意味著需要從用戶態(tài)切換到內(nèi)核態(tài),而這個上下文的切換成本是巨大的。
當使用Pipeline時，它允許多個命令的讀通過一次read()操作，多個命令的應(yīng)答使用一次write()操作，它允許客戶端可以一次發(fā)送多條命令，而不等待上一條命令執(zhí)行的結(jié)果。不僅減少了RTT，同時也減少了IO調(diào)用次數(shù)（IO調(diào)用涉及到用戶態(tài)到內(nèi)核態(tài)之間的切換），最終提升程序的執(zhí)行效率與性能。如下圖：
要支持Pipeline，其實既要服務(wù)端的支持，也要客戶端支持。對于服務(wù)端來說，所需要的是能夠處理一個客戶端通過同一個TCP連接發(fā)來的多個命令，可以理解為，這里將多個命令切分，和處理單個命令一樣，Redis就是這樣處理的。而客戶端，則是要將多個命令緩存起來，緩沖區(qū)滿了就發(fā)送，然后再寫緩沖，最后才處理Redis的應(yīng)答。
三、Pipeline 基本使用及性能比較
下面我們以給10w個set結(jié)構(gòu)分別插入一個整數(shù)值為例，分別使用jedis單個命令插入、jedis使用Pipeline模式進行插入和redisson使用Pipeline模式進行插入以及測試其耗時。
@Slf4j
public class RedisPipelineTestDemo {
public static void main(String[] args) {
//連接redis
Jedis jedis = new Jedis("10.101.17.180", 6379);
//jedis逐一給每個set新增一個value
String zSetKey = "Pipeline-test-set";
int size = 100000;
long begin = System.currentTimeMillis();
for (int i = 0; i < size; i++) {
jedis.sadd(zSetKey + i, "aaa");
}
log.info("Jedis逐一給每個set新增一個value耗時：{}ms", (System.currentTimeMillis() - begin));
//Jedis使用Pipeline模式 Pipeline Pipeline = jedis.Pipelined();
begin = System.currentTimeMillis();
for (int i = 0; i < size; i++) { Pipeline.sadd(zSetKey + i, "bbb");
} Pipeline.sync();
log.info("Jedis Pipeline模式耗時：{}ms", (System.currentTimeMillis() - begin));
//Redisson使用Pipeline模式
Config config = new Config();
config.useSingleServer().setAddress("redis://10.101.17.180:6379");
RedissonClient redisson = Redisson.create(config);
RBatch redisBatch = redisson.createBatch();
begin = System.currentTimeMillis();
for (int i = 0; i < size; i++) {
redisBatch.getSet(zSetKey + i).addAsync("ccc");
}
redisBatch.execute();
log.info("Redisson Pipeline模式耗時：{}ms", (System.currentTimeMillis() - begin));
//關(guān)閉 Pipeline.close();
jedis.close();
redisson.shutdown();
}
}測試結(jié)果如下:
Jedis逐一給每個set新增一個value耗時：162655ms
Jedis Pipeline模式耗時：504ms
Redisson Pipeline模式耗時：1399ms
我們發(fā)現(xiàn)使用Pipeline模式對應(yīng)的性能會明顯好于單個命令執(zhí)行的情況。
四、項目中實際應(yīng)用
在實際使用過程中有這樣一個場景，很多應(yīng)用在節(jié)假日的時候需要更新應(yīng)用圖標樣式，在運營進行后臺配置的時候, 可以根據(jù)圈選的用戶標簽預先計算出單個用戶需要下發(fā)的圖標樣式并存儲在Redis里面，從而提升性能，這里就涉及Redis的批量操作問題，業(yè)務(wù)流程如下：
為了提升Redis操作性能，我們決定使用Redis Pipelining機制進行批量執(zhí)行。
4.1 Redis 客戶端對比
針對Java技術(shù)棧而言，目前Redis使用較多的客戶端為Jedis、Lettuce和Redisson。
目前項目主要是基于SpringBoot開發(fā)，針對Redis，其默認的客戶端為Lettuce，所以我們基于Lettuce客戶端進行分析。
4.2 Spring環(huán)境下Lettuce客戶端對Pipeline的實現(xiàn)
在Spring環(huán)境下，使用Redis的Pipeline也是很簡單的。spring-data-redis提供了
StringRedisTemplate簡化了對Redis的操作, 只需要調(diào)用StringRedisTemplate的executePipelined方法就可以了，但是在參數(shù)中提供了兩種回調(diào)方式：SessionCallback和RedisCallback。
兩種使用方式如下(這里以操作set結(jié)構(gòu)為例)：
RedisCallback的使用方式：
public void testRedisCallback() {
List<Integer> ids= Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
Integer contentId = 1;
redisTemplate.executePipelined(new InsertPipelineExecutionA(ids, contentId));
}
@AllArgsConstructor
private static class InsertPipelineExecutionA implements RedisCallback<Void> {
private final List<Integer> ids;
private final Integer contentId;
@Override
public Void doInRedis(RedisConnection connection) DataAccessException {
RedisSetCommands redisSetCommands = connection.setCommands();
ids.forEach(id-> {
String redisKey = "aaa:" + id;
String value = String.valueOf(contentId);
redisSetCommands.sAdd(redisKey.getBytes(), value.getBytes());
});
return null;
}
}SessionCallback的使用方式：
public void testSessionCallback() {
List<Integer> ids= Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
Integer contentId = 1;
redisTemplate.executePipelined(new InsertPipelineExecutionB(ids, contentId));
}
@AllArgsConstructor
private static class InsertPipelineExecutionB implements SessionCallback<Void> {
private final List<Integer> ids;
private final Integer contentId;
@Override
public <K, V> Void execute(RedisOperations<K, V> operations) throws DataAccessException {
SetOperations<String, String> setOperations = (SetOperations<String, String>) operations.opsForSet();
ids.forEach(id-> {
String redisKey = "aaa:" + id;
String value = String.valueOf(contentId);
setOperations.add(redisKey, value);
});
return null;
}
}4.3 RedisCallBack和SessionCallback之間的比較
1、RedisCallBack和SessionCallback都可以實現(xiàn)回調(diào)，通過它們可以在同一條連接中一次執(zhí)行多個redis命令。
2、RedisCallback使用的是原生
RedisConnection，用起來比較麻煩，比如上面執(zhí)行set的add操作，key和value需要進行轉(zhuǎn)換，可讀性差，但原生api提供的功能比較齊全。
3、SessionCalback提供了良好的封裝，可以優(yōu)先選擇使用這種回調(diào)方式。
最終的代碼實現(xiàn)如下：
public void executeB(List<Integer> userIds, Integer iconId) {
redisTemplate.executePipelined(new InsertPipelineExecution(userIds, iconId));
}
@AllArgsConstructor
private static class InsertPipelineExecution implements SessionCallback<Void> {
private final List<Integer> userIds;
private final Integer iconId;
@Override
public <K, V> Void execute(RedisOperations<K, V> operations) throws DataAccessException {
SetOperations<String, String> setOperations = (SetOperations<String, String>) operations.opsForSet();
userIds.forEach(userId -> {
String redisKey = "aaa:" + userId;
String value = String.valueOf(iconId);
setOperations.add(redisKey, value);
});
return null;
}
}4.4 源碼分析
那么為什么使用Pipeline方式會對性能有較大提升呢，我們現(xiàn)在從源碼入手著重分析一下：
4.4.1 Pipeline方式下獲取連接相關(guān)原理分析：
@Override
public List<Object> executePipelined(SessionCallback<?> session, @Nullable RedisSerializer<?> resultSerializer) {
Assert.isTrue(initialized, "template not initialized; call afterPropertiesSet() before using it");
Assert.notNull(session, "Callback object must not be null");
//1. 獲取對應(yīng)的Redis連接工廠
RedisConnectionFactory factory = getRequiredConnectionFactory();
//2. 綁定連接過程
RedisConnectionUtils.bindConnection(factory, enableTransactionSupport);
try {
//3. 執(zhí)行命令流程, 這里請求參數(shù)為RedisCallback, 里面有對應(yīng)的回調(diào)操作
return execute((RedisCallback<List<Object>>) connection -> {
//具體的回調(diào)邏輯
connection.openPipeline();
boolean PipelinedClosed = false;
try {
//執(zhí)行命令
Object result = executeSession(session);
if (result != null) {
throw new InvalidDataAccessApiUsageException(
"Callback cannot return a non-null value as it gets overwritten by the Pipeline");
}
List<Object> closePipeline = connection.closePipeline(); PipelinedClosed = true;
return deserializeMixedResults(closePipeline, resultSerializer, hashKeySerializer, hashValueSerializer);
} finally {
if (!PipelinedClosed) {
connection.closePipeline();
}
}
});
} finally {
RedisConnectionUtils.unbindConnection(factory);
}
}① 獲取對應(yīng)的Redis連接工廠，這里要使用Pipeline特性需要使用
LettuceConnectionFactory方式，這里獲取的連接工廠就是LettuceConnectionFactory。
② 綁定連接過程，具體指的是將當前連接綁定到當前線程上面， 核心方法為：doGetConnection。
public static RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind,
boolean enableTransactionSupport) {
Assert.notNull(factory, "No RedisConnectionFactory specified");
//核心類，有緩存作用，下次可以從這里獲取已經(jīng)存在的連接
RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory);
//如果connHolder不為null, 則獲取已經(jīng)存在的連接, 提升性能
if (connHolder != null) {
if (enableTransactionSupport) {
potentiallyRegisterTransactionSynchronisation(connHolder, factory);
}
return connHolder.getConnection();
}
......
//第一次獲取連接，需要從Redis連接工廠獲取連接
RedisConnection conn = factory.getConnection();
//bind = true 執(zhí)行綁定
if (bind) {
RedisConnection connectionToBind = conn;
......
connHolder = new RedisConnectionHolder(connectionToBind);
//綁定核心代碼: 將獲取的連接和當前線程綁定起來
TransactionSynchronizationManager.bindResource(factory, connHolder);
......
return connHolder.getConnection();
}
return conn;
}里面有個核心類RedisConnectionHolder，我們看一下
RedisConnectionHolder connHolder =
(RedisConnectionHolder)
TransactionSynchronizationManager.getResource(factory);
@Nullable
public static Object getResource(Object key) {
Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
Object value = doGetResource(actualKey);
if (value != null && logger.isTraceEnabled()) {
logger.trace("Retrieved value [" + value + "] for key [" + actualKey + "] bound to thread [" +
Thread.currentThread().getName() + "]");
}
return value;
}里面有一個核心方法doGetResource
(actualKey)，大家很容易猜測這里涉及到一個map結(jié)構(gòu)，如果我們看源碼，也確實是這樣一個結(jié)構(gòu)。
@Nullable
private static Object doGetResource(Object actualKey) {
Map<Object, Object> map = resources.get();
if (map == null) {
return null;
}
Object value = map.get(actualKey);
// Transparently remove ResourceHolder that was marked as void...
if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) {
map.remove(actualKey);
// Remove entire ThreadLocal if empty...
if (map.isEmpty()) {
resources.remove();
}
value = null;
}
return value;
}resources是一個ThreadLocal類型，這里會涉及到根據(jù)RedisConnectionFactory獲取到連接connection的邏輯,如果下一次是同一個actualKey，那么就直接使用已經(jīng)存在的連接，而不需要新建一個連接。第一次這里map為null，就直接返回了，然后回到doGetConnection方法，由于這里bind為true，我們會執(zhí)行TransactionSynchronizationManager.bindResource(factory, connHolder);，也就是將連接和當前線程綁定了起來。
public static void bindResource(Object key, Object value) throws IllegalStateException {
Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
Assert.notNull(value, "Value must not be null");
Map<Object, Object> map = resources.get();
// set ThreadLocal Map if none found
if (map == null) {
map = new HashMap<>();
resources.set(map);
}
Object oldValue = map.put(actualKey, value);
......
}③ 我們回到executePipelined，在獲取到連接工廠，將連接和當前線程綁定起來以后，就開始需要正式去執(zhí)行命令了, 這里會調(diào)用execute方法
@Override
@Nullable
public <T> T execute(RedisCallback<T> action) {
return execute(action, isExposeConnection());
}這里我們注意到execute方法的入?yún)镽edisCallback<T>action，RedisCallback對應(yīng)的doInRedis操作如下,這里在后面的調(diào)用過程中會涉及到回調(diào)。
connection.openPipeline();
boolean PipelinedClosed = false;
try {
Object result = executeSession(session);
if (result != null) {
throw new InvalidDataAccessApiUsageException(
"Callback cannot return a non-null value as it gets overwritten by the Pipeline");
}
List<Object> closePipeline = connection.closePipeline(); PipelinedClosed = true;
return deserializeMixedResults(closePipeline, resultSerializer, hashKeySerializer, hashValueSerializer);
} finally {
if (!PipelinedClosed) {
connection.closePipeline();
}
}我們再來看execute(action,
isExposeConnection())方法，這里最終會調(diào)用
<T>execute(RedisCallback<T>action, boolean exposeConnection, boolean Pipeline)方法。
@Nullable
public <T> T execute(RedisCallback<T> action, boolean exposeConnection, boolean Pipeline) {
Assert.isTrue(initialized, "template not initialized; call afterPropertiesSet() before using it");
Assert.notNull(action, "Callback object must not be null");
//獲取對應(yīng)的連接工廠
RedisConnectionFactory factory = getRequiredConnectionFactory();
RedisConnection conn = null;
try {
if (enableTransactionSupport) {
// only bind resources in case of potential transaction synchronization
conn = RedisConnectionUtils.bindConnection(factory, enableTransactionSupport);
} else {
//獲取對應(yīng)的連接(enableTransactionSupport=false)
conn = RedisConnectionUtils.getConnection(factory);
}
boolean existingConnection = TransactionSynchronizationManager.hasResource(factory);
RedisConnection connToUse = preProcessConnection(conn, existingConnection);
boolean PipelineStatus = connToUse.isPipelined();
if (Pipeline && !PipelineStatus) {
connToUse.openPipeline();
}
RedisConnection connToExpose = (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
//核心方法，這里就開始執(zhí)行回調(diào)操作
T result = action.doInRedis(connToExpose);
// close Pipeline
if (Pipeline && !PipelineStatus) {
connToUse.closePipeline();
}
// TODO: any other connection processing?
return postProcessResult(result, connToUse, existingConnection);
} finally {
RedisConnectionUtils.releaseConnection(conn, factory, enableTransactionSupport);
}
}我們看到這里最開始也是獲取對應(yīng)的連接工廠，然后獲取對應(yīng)的連接
(enableTransactionSupport=false)，具體調(diào)用是
RedisConnectionUtils.getConnection(factory)方法，最終會調(diào)用
RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind, boolean enableTransactionSupport)，此時bind為false
public static RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind,
boolean enableTransactionSupport) {
Assert.notNull(factory, "No RedisConnectionFactory specified");
//直接獲取與當前線程綁定的Redis連接
RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory);
if (connHolder != null) {
if (enableTransactionSupport) {
potentiallyRegisterTransactionSynchronisation(connHolder, factory);
}
return connHolder.getConnection();
}
......
return conn;
}前面我們分析過一次，這里調(diào)用
RedisConnectionHolder connHolder =
(RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory);會獲取到之前和當前線程綁定的Redis，而不會新創(chuàng)建一個連接。
然后會去執(zhí)行T result = action.
doInRedis(connToExpose)，這里的action為RedisCallback，執(zhí)行doInRedis為：
//開啟Pipeline功能
connection.openPipeline();
boolean PipelinedClosed = false;
try {
//執(zhí)行Redis命令
Object result = executeSession(session);
if (result != null) {
throw new InvalidDataAccessApiUsageException(
"Callback cannot return a non-null value as it gets overwritten by the Pipeline");
}
List<Object> closePipeline = connection.closePipeline(); PipelinedClosed = true;
return deserializeMixedResults(closePipeline, resultSerializer, hashKeySerializer, hashValueSerializer);
} finally {
if (!PipelinedClosed) {
connection.closePipeline();
}
}這里最開始會開啟Pipeline功能，然后執(zhí)行
Object result = executeSession(session);
private Object executeSession(SessionCallback<?> session) {
return session.execute(this);
}這里會調(diào)用我們自定義的execute方法
@AllArgsConstructor
private static class InsertPipelineExecution implements SessionCallback<Void> {
private final List<Integer> userIds;
private final Integer iconId;
@Override
public <K, V> Void execute(RedisOperations<K, V> operations) throws DataAccessException {
SetOperations<String, String> setOperations = (SetOperations<String, String>) operations.opsForSet();
userIds.forEach(userId -> {
String redisKey = "aaa:" + userId;
String value = String.valueOf(iconId);
setOperations.add(redisKey, value);
});
return null;
}
}進入到foreach循環(huán)，執(zhí)行DefaultSetOperations的add方法。
@Override
public Long add(K key, V... values) {
byte[] rawKey = rawKey(key);
byte[][] rawValues = rawValues((Object[]) values);
//這里的connection.sAdd是后續(xù)回調(diào)要執(zhí)行的方法
return execute(connection -> connection.sAdd(rawKey, rawValues), true);
}這里會繼續(xù)執(zhí)行redisTemplate的execute方法，里面最終會調(diào)用我們之前分析過的<T>T execute(RedisCallback<T>action, boolean exposeConnection, boolean Pipeline)方法。
@Nullable
public <T> T execute(RedisCallback<T> action, boolean exposeConnection, boolean Pipeline) {
Assert.isTrue(initialized, "template not initialized; call afterPropertiesSet() before using it");
Assert.notNull(action, "Callback object must not be null");
RedisConnectionFactory factory = getRequiredConnectionFactory();
RedisConnection conn = null;
try {
......
//再次執(zhí)行回調(diào)方法，這里執(zhí)行的Redis基本數(shù)據(jù)結(jié)構(gòu)對應(yīng)的操作命令
T result = action.doInRedis(connToExpose);
......
// TODO: any other connection processing?
return postProcessResult(result, connToUse, existingConnection);
} finally {
RedisConnectionUtils.releaseConnection(conn, factory, enableTransactionSupport);
}
}這里會繼續(xù)執(zhí)行T result =
action.doInRedis(connToExpose);，這里其實執(zhí)行的doInRedis方法為：
connection -> connection.sAdd(rawKey, rawValues)4.4.2 Pipeline方式下執(zhí)行命令的流程分析:
① 接著上面的流程分析，這里的sAdd方法實際調(diào)用的是DefaultStringRedisConnection的sAdd方法
@Override
public Long sAdd(byte[] key, byte[]... values) {
return convertAndReturn(delegate.sAdd(key, values), identityConverter);
}② 這里會進一步調(diào)用
DefaultedRedisConnection的sAdd方法
@Override
@Deprecated
default Long sAdd(byte[] key, byte[]... values) {
return setCommands().sAdd(key, values);
}③ 接著調(diào)用LettuceSetCommands的sAdd方法
@Override
public Long sAdd(byte[] key, byte[]... values) {
Assert.notNull(key, "Key must not be null!");
Assert.notNull(values, "Values must not be null!");
Assert.noNullElements(values, "Values must not contain null elements!");
try {
// 如果開啟了 Pipelined 模式，獲取的是 異步連接，進行異步操作
if (isPipelined()) { Pipeline(connection.newLettuceResult(getAsyncConnection().sadd(key, values)));
return null;
}
if (isQueueing()) {
transaction(connection.newLettuceResult(getAsyncConnection().sadd(key, values)));
return null;
}
//常規(guī)模式下，使用的是同步操作
return getConnection().sadd(key, values);
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}這里我們開啟了Pipeline, 實際會調(diào)用
Pipeline(connection.newLettuceResult(getAsyncConnection().sadd(key, values))); 也就是獲取異步連接getAsyncConnection，然后進行異步操作sadd，而常規(guī)模式下，使用的是同步操作，所以在Pipeline模式下，執(zhí)行效率更高。
從上面的獲取連接和具體命令執(zhí)行相關(guān)源碼分析可以得出使用Lettuce客戶端Pipeline模式高效的根本原因：
- 普通模式下，每執(zhí)行一個命令都需要先打開一個連接，命令執(zhí)行完畢以后又需要關(guān)閉這個連接，執(zhí)行下一個命令時，又需要經(jīng)過連接打開和關(guān)閉的流程；而Pipeline的所有命令的執(zhí)行只需要經(jīng)過一次連接打開和關(guān)閉。
- 普通模式下命令的執(zhí)行是同步阻塞模式，而Pipeline模式下命令的執(zhí)行是異步非阻塞模式。
五、項目中遇到的坑
前面介紹了涉及到批量操作，可以使用Redis Pipelining機制，那是不是任何批量操作相關(guān)的場景都可以使用呢，比如list類型數(shù)據(jù)的批量移除操作，我們的代碼最開始是這么寫的：
public void deleteSet(String updateKey, Set<Integer> userIds) {
if (CollectionUtils.isEmpty(userIds)) {
return;
}
redisTemplate.executePipelined(new DeleteListCallBack(userIds, updateKey));
}
@AllArgsConstructor
private static class DeleteListCallBack implements SessionCallback<Object> {
private Set<Integer> userIds;
private String updateKey;
@Override
public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
ListOperations<String, String> listOperations = (ListOperations<String, String>) operations.opsForList();
userIds.forEach(userId -> listOperations.remove(updateKey, 1, userId.toString()));
return null;
}
}在數(shù)據(jù)量比較小的時候沒有出現(xiàn)問題，直到有一條收到了Redis的內(nèi)存和cpu利用率的告警消息，我們發(fā)現(xiàn)這么使用是有問題的，核心原因在于list的lrem操作的時間復雜度是O(N+M)，其中N是list的長度, M是要移除的元素的個數(shù)，而我們這里還是一個一個移除的，當然會導致Redis數(shù)據(jù)積壓和cpu每秒ops升高導致cpu利用率飚高。也就是說，即使使用Pipeline進行批量操作，但是由于單次操作很耗時，是會導致整個Redis出現(xiàn)問題的。
后面我們進行了優(yōu)化，選用了list的ltrim命令，一次命令執(zhí)行批量remove操作：
public void deleteSet(String updateKey, Set<Integer> deviceIds) {
if (CollectionUtils.isEmpty(deviceIds)) {
return;
}
int maxSize = 10000;
redisTemplate.opsForList().trim(updateKey, maxSize + 1, -1);
}由于ltrim本身的時間復雜度為O(M), 其中M要移除的元素的個數(shù)，相比于原始方案的lrem，效率提升很多，可以不需要使用Redis Pipeline，優(yōu)化結(jié)果使得Redis內(nèi)存利用率和cpu利用率都極大程度得到緩解。
六、Redisson 對 Redis Pipeline 特性支持
在redisson官方文檔中額外特性介紹中有說到批量命令執(zhí)行這個特性， 也就是多個命令在一次網(wǎng)絡(luò)調(diào)用中集中發(fā)送，該特性是RBatch這個類支持的，從這個類的描述來看，主要是為Redis Pipeline這個特性服務(wù)的，并且主要是通過隊列和異步實現(xiàn)的。
/**
* Interface for using Redis Pipeline feature.
* <p>
* All method invocations on objects got through this interface
* are batched to separate queue and could be executed later
* with <code>execute()</code> or <code>executeAsync()</code> methods.
*
*
* @author Nikita Koksharov
*
*/
public interface RBatch {
/**
* Returns stream instance by <code>name</code>
*
* @param <K> type of key
* @param <V> type of value
* @param name of stream
* @return RStream object
*/
<K, V> RStreamAsync<K, V> getStream(String name);
/**
* Returns stream instance by <code>name</code>
* using provided <code>codec</code> for entries.
*
* @param <K> type of key
* @param <V> type of value
* @param name - name of stream
* @param codec - codec for entry
* @return RStream object
*/
<K, V> RStreamAsync<K, V> getStream(String name, Codec codec);
......
/**
* Returns list instance by name.
*
* @param <V> type of object
* @param name - name of object
* @return List object
*/
<V> RListAsync<V> getList(String name);
<V> RListAsync<V> getList(String name, Codec codec);
......
/**
* Executes all operations accumulated during async methods invocations.
* <p>
* If cluster configuration used then operations are grouped by slot ids
* and may be executed on different servers. Thus command execution order could be changed
*
* @return List with result object for each command
* @throws RedisException in case of any error
*
*/
BatchResult<?> execute() throws RedisException;
/**
* Executes all operations accumulated during async methods invocations asynchronously.
* <p>
* In cluster configurations operations grouped by slot ids
* so may be executed on different servers. Thus command execution order could be changed
*
* @return List with result object for each command
*/
RFuture<BatchResult<?>> executeAsync();
/**
* Discard batched commands and release allocated buffers used for parameters encoding.
*/
void discard();
/**
* Discard batched commands and release allocated buffers used for parameters encoding.
*
* @return void
*/
RFuture<Void> discardAsync();
}簡單的測試代碼如下：
@Slf4j
public class RedisPipelineTest {
public static void main(String[] args) {
//Redisson使用Pipeline模式
Config config = new Config();
config.useSingleServer().setAddress("redis://xx.xx.xx.xx:6379");
RedissonClient redisson = Redisson.create(config);
RBatch redisBatch = redisson.createBatch();
int size = 100000;
String zSetKey = "Pipeline-test-set";
long begin = System.currentTimeMillis();
//將命令放入隊列中
for (int i = 0; i < size; i++) {
redisBatch.getSet(zSetKey + i).addAsync("ccc");
}
//批量執(zhí)行命令
redisBatch.execute();
log.info("Redisson Pipeline模式耗時：{}ms", (System.currentTimeMillis() - begin));
//關(guān)閉
redisson.shutdown();
}
}核心方法分析：
1.建Redisson客戶端RedissonClient redisson = redisson.create(config), 該方法最終會調(diào)用Reddison的構(gòu)造方法Redisson(Config config)。
protected Redisson(Config config) {
this.config = config;
Config configCopy = new Config(config);
connectionManager = ConfigSupport.createConnectionManager(configCopy);
RedissonObjectBuilder objectBuilder = null;
if (config.isReferenceEnabled()) {
objectBuilder = new RedissonObjectBuilder(this);
}
//新建異步命令執(zhí)行器
commandExecutor = new CommandSyncService(connectionManager, objectBuilder);
//執(zhí)行刪除超時任務(wù)的定時器
evictionScheduler = new EvictionScheduler(commandExecutor);
writeBehindService = new WriteBehindService(commandExecutor);
}該構(gòu)造方法中會新建異步命名執(zhí)行器CommandAsyncExecutor commandExecutor和用戶刪除超時任務(wù)的EvictionScheduler evictionScheduler。
2.創(chuàng)建RBatch實例RBatch redisBatch = redisson.createBatch()， 該方法會使用到步驟1中的commandExecutor和evictionScheduler實例對象。
@Override
public RBatch createBatch(BatchOptions options) {
return new RedissonBatch(evictionScheduler, commandExecutor, options);
}
public RedissonBatch(EvictionScheduler evictionScheduler, CommandAsyncExecutor executor, BatchOptions options) {
this.executorService = new CommandBatchService(executor, options);
this.evictionScheduler = evictionScheduler;
}其中的options對象會影響后面批量執(zhí)行命令的流程。
3. 異步給set集合添加元素的操作addAsync，這里會具體調(diào)用RedissonSet的addAsync方法
@Override
public RFuture<Boolean> addAsync(V e) {
String name = getRawName(e);
return commandExecutor.writeAsync(name, codec, RedisCommands.SADD_SINGLE, name, encode(e));
}（1）接著調(diào)用CommandAsyncExecutor的異步寫入方法writeAsync。
@Override
public <T, R> RFuture<R> writeAsync(String key, Codec codec, RedisCommand<T> command, Object... params) {
RPromise<R> mainPromise = createPromise();
NodeSource source = getNodeSource(key);
async(false, source, codec, command, params, mainPromise, false);
return mainPromise;
}（2） 接著調(diào)用批量命令執(zhí)行器
CommandBatchService的異步發(fā)送命令。
@Override
public <V, R> void async(boolean readOnlyMode, NodeSource nodeSource,
Codec codec, RedisCommand<V> command, Object[] params, RPromise<R> mainPromise, boolean ignoreRedirect) {
if (isRedisBasedQueue()) {
boolean isReadOnly = options.getExecutionMode() == ExecutionMode.REDIS_READ_ATOMIC;
RedisExecutor<V, R> executor = new RedisQueuedBatchExecutor<>(isReadOnly, nodeSource, codec, command, params, mainPromise,
false, connectionManager, objectBuilder, commands, connections, options, index, executed, latch, referenceType);
executor.execute();
} else {
//執(zhí)行分支
RedisExecutor<V, R> executor = new RedisBatchExecutor<>(readOnlyMode, nodeSource, codec, command, params, mainPromise,
false, connectionManager, objectBuilder, commands, options, index, executed, referenceType);
executor.execute();
}
}（3） 接著調(diào)用了RedisBatchExecutor.
execute方法和BaseRedisBatchExecutor.
addBatchCommandData方法。
@Override
public void execute() {
addBatchCommandData(params);
}
protected final void addBatchCommandData(Object[] batchParams) {
MasterSlaveEntry msEntry = getEntry(source);
Entry entry = commands.get(msEntry);
if (entry == null) {
entry = new Entry();
Entry oldEntry = commands.putIfAbsent(msEntry, entry);
if (oldEntry != null) {
entry = oldEntry;
}
}
if (!readOnlyMode) {
entry.setReadOnlyMode(false);
}
Codec codecToUse = getCodec(codec);
BatchCommandData<V, R> commandData = new BatchCommandData<V, R>(mainPromise, codecToUse, command, batchParams, index.incrementAndGet());
entry.getCommands().add(commandData);
}這里的commands以主節(jié)點為KEY,以待發(fā)送命令隊列列表為VALUE(Entry)，保存一個MAP.然后會把命令都添加到entry的commands命令隊列中， Entry結(jié)構(gòu)如下面代碼所示。
public static class Entry {
Deque<BatchCommandData<?, ?>> commands = new LinkedBlockingDeque<>();
volatile boolean readOnlyMode = true;
public Deque<BatchCommandData<?, ?>> getCommands() {
return commands;
}
public void setReadOnlyMode(boolean readOnlyMode) {
this.readOnlyMode = readOnlyMode;
}
public boolean isReadOnlyMode() {
return readOnlyMode;
}
public void clearErrors() {
for (BatchCommandData<?, ?> commandEntry : commands) {
commandEntry.clearError();
}
}
}4. 批量執(zhí)行命令redisBatch.execute()，這里會最終調(diào)用CommandBatchService的executeAsync方法，該方法完整代碼如下，我們下面來逐一進行拆解。
public RFuture<BatchResult<?>> executeAsync() {
......
RPromise<BatchResult<?>> promise = new RedissonPromise<>();
RPromise<Void> voidPromise = new RedissonPromise<Void>();
if (this.options.isSkipResult()
&& this.options.getSyncSlaves() == 0) {
......
} else {
//這里是對異步執(zhí)行結(jié)果進行處理,可以先忽略, 后面會詳細講，先關(guān)注批量執(zhí)行命令的邏輯
voidPromise.onComplete((res, ex) -> {
......
});
}
AtomicInteger slots = new AtomicInteger(commands.size());
......
//真正執(zhí)行的代碼入口，批量執(zhí)行命令
for (Map.Entry<MasterSlaveEntry, Entry> e : commands.entrySet()) {
RedisCommonBatchExecutor executor = new RedisCommonBatchExecutor(new NodeSource(e.getKey()), voidPromise,
connectionManager, this.options, e.getValue(), slots, referenceType);
executor.execute();
}
return promise;
}里面會用到我們在3.3步驟所生成的commands實例。
（1）接著調(diào)用了基類RedisExecutor的execute方法
public void execute() {
......
connectionFuture.onComplete((connection, e) -> {
if (connectionFuture.isCancelled()) {
connectionManager.getShutdownLatch().release();
return;
}
if (!connectionFuture.isSuccess()) {
connectionManager.getShutdownLatch().release();
exception = convertException(connectionFuture);
return;
}
//調(diào)用RedisCommonBatchExecutor的sendCommand方法, 里面會將多個命令放到一個List<CommandData<?, ?>> list列表里面
sendCommand(attemptPromise, connection);
writeFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
checkWriteFuture(writeFuture, attemptPromise, connection);
}
});
});
......
}（2）接著調(diào)用
RedisCommonBatchExecutor的sendCommand方法，里面會將多個命令放到一個List<commanddata> list列表里面。
@Override
protected void sendCommand(RPromise<Void> attemptPromise, RedisConnection connection) {
boolean isAtomic = options.getExecutionMode() != ExecutionMode.IN_MEMORY;
boolean isQueued = options.getExecutionMode() == ExecutionMode.REDIS_READ_ATOMIC
|| options.getExecutionMode() == ExecutionMode.REDIS_WRITE_ATOMIC;
//將多個命令放到一個List<CommandData<?, ?>> list列表里面
List<CommandData<?, ?>> list = new ArrayList<>(entry.getCommands().size());
if (source.getRedirect() == Redirect.ASK) {
RPromise<Void> promise = new RedissonPromise<Void>();
list.add(new CommandData<Void, Void>(promise, StringCodec.INSTANCE, RedisCommands.ASKING, new Object[] {}));
}
for (CommandData<?, ?> c : entry.getCommands()) {
if ((c.getPromise().isCancelled() || c.getPromise().isSuccess())
&& !isWaitCommand(c)
&& !isAtomic) {
// skip command
continue;
}
list.add(c);
}
......
//調(diào)用RedisConnection的send方法，將命令一次性發(fā)到Redis服務(wù)器端
writeFuture = connection.send(new CommandsData(attemptPromise, list, options.isSkipResult(), isAtomic, isQueued, options.getSyncSlaves() > 0));
}（3）接著調(diào)用RedisConnection的send方法，通過Netty通信發(fā)送命令到Redis服務(wù)器端執(zhí)行，這里也驗證了Redisson客戶端底層是采用Netty進行通信的。
public ChannelFuture send(CommandsData data) {
return channel.writeAndFlush(data);
}5. 接收返回結(jié)果，這里主要是監(jiān)聽事件是否完成，然后組裝返回結(jié)果, 核心方法是步驟4提到的CommandBatchService的executeAsync方法,里面會對返回結(jié)果進行監(jiān)聽和處理, 核心代碼如下：
public RFuture<BatchResult<?>> executeAsync() {
......
RPromise<BatchResult<?>> promise = new RedissonPromise<>();
RPromise<Void> voidPromise = new RedissonPromise<Void>();
if (this.options.isSkipResult()
&& this.options.getSyncSlaves() == 0) {
......
} else {
voidPromise.onComplete((res, ex) -> {
//對返回結(jié)果的處理
executed.set(true);
......
List<Object> responses = new ArrayList<Object>(entries.size());
int syncedSlaves = 0;
for (BatchCommandData<?, ?> commandEntry : entries) {
if (isWaitCommand(commandEntry)) {
syncedSlaves = (Integer) commandEntry.getPromise().getNow();
} else if (!commandEntry.getCommand().getName().equals(RedisCommands.MULTI.getName())
&& !commandEntry.getCommand().getName().equals(RedisCommands.EXEC.getName())
&& !this.options.isSkipResult()) {
......
//獲取單個命令的執(zhí)行結(jié)果
Object entryResult = commandEntry.getPromise().getNow();
......
//將單個命令執(zhí)行結(jié)果放到List中
responses.add(entryResult);
}
}
BatchResult<Object> result = new BatchResult<Object>(responses, syncedSlaves);
promise.trySuccess(result);
......
});
}
......
return promise;
}這里會把單個命令的執(zhí)行結(jié)果放到responses里面，最終返回RPromise<batchresult>promise。
從上面的分析來看，Redisson客戶端對Redis Pipeline的支持也是從多個命令在一次網(wǎng)絡(luò)通信中執(zhí)行和異步處理來實現(xiàn)的。
七、總結(jié)
Redis提供了Pipelining進行批量操作的高級特性，極大地提高了部分數(shù)據(jù)類型沒有批量執(zhí)行命令導致的執(zhí)行耗時而引起的性能問題，但是我們在使用的過程中需要考慮Pipeline操作中單個命令執(zhí)行的耗時問題，否則帶來的效果可能適得其反。最后擴展分析了Redisson客戶端對Redis Pipeline特性的支持原理，可以與Lettuce客戶端對Redis Pipeline支持原理進行比較，加深Pipeline在不同Redis客戶端實現(xiàn)方式的理解。
