...
作者：vivo 互联网服务器团队-Wang Fei
Redis是一种基于客户端-服务端模型以及请求/响应的TCP服务。在遇到批处理命令执行时，Redis提供了Pipelining(管道)来提升批处理性能。本文结合实践分析了Spring Boot框架下Redis的Lettuce客户端和Redisson客户端对Pipeline特性的支持原理，并针对实践过程中遇到的问题进行了分析，可以帮助开发者了解不同客户端对Pipeline支持原理及避免实际使用中出现问题。
Redis 已经提供了像 mget 、mset 这种批量的命令，但是某些操作根本就不支持或没有批量的操作，从而与 Redis 高性能背道而驰。为此, Redis基于管道机制，提供Redis Pipeline新特性。Redis Pipeline是一种通过一次性发送多条命令并在执行完后一次性将结果返回，从而减少客户端与redis的通信次数来实现降低往返延时时间提升操作性能的技术。目前，Redis Pipeline是被很多个版本的Redis 客户端所支持的。
2.1 Redis单个命令执行基本步骤
Redis是一种基于客户端-服务端模型以及请求/响应的TCP服务。一次Redis客户端发起的请求，经过服务端的响应后，大致会经历如下的步骤：
客户端发起一个（查询/插入）请求，并监听socket返回，通常情况都是阻塞模式等待Redis服务器的响应。
服务端处理命令，并且返回处理结果给客户端。
客户端接收到服务的返回结果，程序从阻塞代码处返回。
2.2 RTT 时间
Redis客户端和服务端之间通过网络连接进行数据传输，数据包从客户端到达服务器，并从服务器返回数据回复客户端的时间被称之为RTT(Round Trip Time - 往返时间)。我们可以很容易就意识到，Redis在连续请求服务端时，如果RTT时间为250ms, 即使Redis每秒能处理100k请求，但也会因为网络传输花费大量时间，导致每秒最多也只能处理4个请求，导致整体性能的下降。
2.3 Redis Pipeline
为了提升效率，这时候Pipeline出现了。Pipelining不仅仅能够降低RRT，实际上它极大的提升了单次执行的操作数。这是因为如果不使用Pipelining，那么每次执行单个命令,从访问数据的结构和服务端产生应答的角度，它的成本是很低的。但是从执行网络IO的角度，它的成本其实是很高的。其中涉及到read()和write()的系统调用，这意味着需要从用户态切换到内核态,而这个上下文的切换成本是巨大的。
当使用Pipeline时，它允许多个命令的读通过一次read()操作，多个命令的应答使用一次write()操作，它允许客户端可以一次发送多条命令，而不等待上一条命令执行的结果。不仅减少了RTT，同时也减少了IO调用次数（IO调用涉及到用户态到内核态之间的切换），最终提升程序的执行效率与性能。如下图：
要支持Pipeline，其实既要服务端的支持，也要客户端支持。对于服务端来说，所需要的是能够处理一个客户端通过同一个TCP连接发来的多个命令，可以理解为，这里将多个命令切分，和处理单个命令一样，Redis就是这样处理的。而客户端，则是要将多个命令缓存起来，缓冲区满了就发送，然后再写缓冲，最后才处理Redis的应答。
下面我们以给10w个set结构分别插入一个整数值为例，分别使用jedis单个命令插入、jedis使用Pipeline模式进行插入和redisson使用Pipeline模式进行插入以及测试其耗时。
@Slf4jpublic class RedisPipelineTestDemo { public static void main(String[] args) { Jedis jedis = new Jedis("10.101.17.180", 6379); 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)); 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)); 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)); jedis.close(); redisson.shutdown(); }}
测试结果如下:
Jedis逐一给每个set新增一个value耗时：162655ms
Jedis Pipeline模式耗时：504ms
Redisson Pipeline模式耗时：1399ms
我们发现使用Pipeline模式对应的性能会明显好于单个命令执行的情况。
在实际使用过程中有这样一个场景，很多应用在节假日的时候需要更新应用图标样式，在运营进行后台配置的时候, 可以根据圈选的用户标签预先计算出单个用户需要下发的图标样式并存储在Redis里面，从而提升性能，这里就涉及Redis的批量操作问题，业务流程如下：
为了提升Redis操作性能，我们决定使用Redis Pipelining机制进行批量执行。
4.1 Redis 客户端对比
针对Java技术栈而言，目前Redis使用较多的客户端为Jedis、Lettuce和Redisson。
目前项目主要是基于SpringBoot开发，针对Redis，其默认的客户端为Lettuce，所以我们基于Lettuce客户端进行分析。
4.2 Spring环境下Lettuce客户端对Pipeline的实现
在Spring环境下，使用Redis的Pipeline也是很简单的。spring-data-redis提供了
StringRedisTemplate简化了对Redis的操作, 只需要调用StringRedisTemplate的executePipelined方法就可以了，但是在参数中提供了两种回调方式：SessionCallback和RedisCallback。
两种使用方式如下(这里以操作set结构为例)：
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; } }
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都可以实现回调，通过它们可以在同一条连接中一次执行多个redis命令。
2、RedisCallback使用的是原生
RedisConnection，用起来比较麻烦，比如上面执行set的add操作，key和value需要进行转换，可读性差，但原生api提供的功能比较齐全。
3、SessionCalback提供了良好的封装，可以优先选择使用这种回调方式。
最终的代码实现如下：
public void executeB(List<Integer> userIds, Integer iconId) { redisTemplate.executePipelined(new InsertPipelineExecution(userIds, iconId));} @AllArgsConstructorprivate 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方式会对性能有较大提升呢，我们现在从源码入手着重分析一下：
4.4.1 Pipeline方式下获取连接相关原理分析：
@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"); RedisConnectionFactory factory = getRequiredConnectionFactory(); RedisConnectionUtils.bindConnection(factory, enableTransactionSupport); try { return execute((RedisCallback<List<Object>>) connection -> { 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(); } } }); } finally { RedisConnectionUtils.unbindConnection(factory); } }
① 获取对应的Redis连接工厂，这里要使用Pipeline特性需要使用
LettuceConnectionFactory方式，这里获取的连接工厂就是LettuceConnectionFactory。
② 绑定连接过程，具体指的是将当前连接绑定到当前线程上面， 核心方法为：doGetConnection。
public static RedisConnection doGetConnection(RedisConnectionFactory factory, boolean allowCreate, boolean bind, boolean enableTransactionSupport) { Assert.notNull(factory, "No RedisConnectionFactory specified"); RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory); if (connHolder != null) { if (enableTransactionSupport) { potentiallyRegisterTransactionSynchronisation(connHolder, factory); } return connHolder.getConnection(); } ...... RedisConnection conn = factory.getConnection(); 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结构，如果我们看源码，也确实是这样一个结构。
@Nullable private static Object doGetResource(Object actualKey) { Map<Object, Object> map = resources.get(); if (map == null) { return null; } Object value = map.get(actualKey); if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) { map.remove(actualKey); if (map.isEmpty()) { resources.remove(); } value = null; } return value; }
resources是一个ThreadLocal类型，这里会涉及到根据RedisConnectionFactory获取到连接connection的逻辑,如果下一次是同一个actualKey，那么就直接使用已经存在的连接，而不需要新建一个连接。第一次这里map为null，就直接返回了，然后回到doGetConnection方法，由于这里bind为true，我们会执行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(); if (map == null) { map = new HashMap<>(); resources.set(map); } Object oldValue = map.put(actualKey, value); ...... }
③ 我们回到executePipelined，在获取到连接工厂，将连接和当前线程绑定起来以后，就开始需要正式去执行命令了, 这里会调用execute方法
@Override@Nullablepublic <T> T execute(RedisCallback<T> action) { return execute(action, isExposeConnection());}
这里我们注意到execute方法的入参为RedisCallback<T>action，RedisCallback对应的doInRedis操作如下,这里在后面的调用过程中会涉及到回调。
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())方法，这里最终会调用
<T>execute(RedisCallback<T>action, boolean exposeConnection, boolean Pipeline)方法。
@Nullablepublic <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 { if (enableTransactionSupport) { conn = RedisConnectionUtils.bindConnection(factory, enableTransactionSupport); }