Spring-data-redis + Lettuce 如何使用 Pipeline-阿里云开发者社区

Spring-data-redis + Lettuce 如何使用 Pipeline

2022-06-25 1220

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

Redis 开源版，标准版 2GB

云数据库 Tair（兼容Redis），内存型 2GB

简介： Spring-data-redis + Lettuce 如何使用 Pipeline

关于 spring-data-redis 和 lettuce，笔者写过不少文章：

最近，私信还有留言中，网友提到 spring-data-redis 和 lettuce 一起使用，pipeline 通过抓包一看，并没有生效，这个如何配置才能生效呢？

首先，在上面的文章中，我们分析过 Spring-data-redis + Lettuce 的基本原理，在这种环境下 RedisTemplate 使用的连接内部包括：

asyncSharedConn：可以为空，如果开启了连接共享，则不为空，默认是开启的；所有 LettuceConnection 共享的 Redis 连接，对于每个 LettuceConnection 实际上都是同一个连接；用于执行简单命令，因为 Netty 客户端与 Redis 的单处理线程特性，共享同一个连接也是很快的。如果没开启连接共享，则这个字段为空，使用 asyncDedicatedConn 执行命令。
asyncDedicatedConn：私有连接，如果需要保持会话，执行事务，以及 Pipeline 命令，固定连接，则必须使用这个 asyncDedicatedConn 执行 Redis 命令。

execute(RedisCallback)，流程是：

对于 executePipelined(RedisCallback)，如果使用正确的话，会使用 asyncDedicatedConn 私有连接执行。那么怎么算使用正确呢？

需要使用回调的连接进行 Redis 调用，不能直接使用redisTemplate调用，否则 pipeline 不生效：

Pipeline 生效：

List<Object> objects = redisTemplate.executePipelined(new RedisCallback<Object>() {
    @Override
    public Object doInRedis(RedisConnection connection) throws DataAccessException {
        connection.get("test".getBytes());
        connection.get("test2".getBytes());
        return null;
    }
});

Pipeline 不生效：

List<Object> objects = redisTemplate.executePipelined(new RedisCallback<Object>() {
    @Override
    public Object doInRedis(RedisConnection connection) throws DataAccessException {
        redisTemplate.opsForValue().get("test");
        redisTemplate.opsForValue().get("test2");
        return null;
    }
});

这样我们就能使用保证 API 层正确使用 pipeline 了，但是默认配置的情况下， 底层还是没有执行 Pipeline，这是怎么回事呢？

Redis Pipeline 类比 Lettuce 中的 AutoFlushCommands

Redis Pipeline 是 Redis 中的批量操作，它能将一组 Redis 命令进行组装，通过一次传输给 Redis 并返回结果集，大大减少了如果命令时一条条单独传输需要的 RTT 时间（包括 Redis 客户端，Redis 服务端切换系统调用发送接收数据的时间，以及网络传输时间）。

如果原来的命令是这么发送的：

Client -> Server: INCR X\r\n
Server -> Client: 1
Client -> Server: INCR X\r\n
Server -> Client: 2
Client -> Server: INCR X\r\n
Server -> Client: 3
Client -> Server: INCR X\r\n
Server -> Client: 4

那么使用 PIPELINE 之后，命令就是类似于这么发送的

Client -> Server: INCR X\r\nINCR X\r\nINCR X\r\nINCR X\r\n
Server -> Client: 1\r\n2\r\n3\r\n4

我们可以看出，其实它的原理，就是客户端先将所有命令拼接在一起然后本地缓存起来，之后统一发到服务端，服务端执行所有命令之后，统一响应。

Lettuce 的连接有一个 AutoFlushCommands 配置，就是指在这个连接上执行的命令，如果发送到服务端。默认是 false，即收到一个命令就发到服务端一个。如果配置为 false，则将所有命令缓存起来，手动调用 flushCommands 的时候，将缓存的命令一起发到服务端，这样其实就是实现了 Pipeline。

配置 Spring-data-redis + Lettuce 使用 Pipeline

Spring-data-redis 从 2.3.0 版本开始，对于 Lettuce 也兼容了 Pipeline 配置，参考：

我们可以这样配置：

@Bean
public BeanPostProcessor lettuceConnectionFactoryBeanProcessor() {
    return new BeanPostProcessor() {
        @Override
        public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
            //在 LettuceConnectionFactory 这个 Bean 初始化之后，设置 PipeliningFlushPolicy 为 flushOnClose
            if (bean instanceof LettuceConnectionFactory) {
                LettuceConnectionFactory lettuceConnectionFactory = (LettuceConnectionFactory) bean;
                lettuceConnectionFactory.setPipeliningFlushPolicy(LettuceConnection.PipeliningFlushPolicy.flushOnClose());
                //多谢评论区的[孤胆枪手](https://juejin.cn/user/2084329775180605) 的指正，我之前忘了放这个配置了
                lettuceConnectionFactory.setShareNativeConnection(false);
            }
            return bean;
        }
    };
}

注意这里将 shareNativeConnection 设置为 false。本来基于 Lettuce 的 RedisTemplate 中大部分请求都可以通过共享连接使用同一个连接，关闭的话每次都获取的是独占连接。这种情况下我们要注意使用连接池（防止每次创建新连接），同时连接池需要大于可能的并发线程个数防止阻塞等待连接。

为啥要关闭这个共享链接呢，参考源码：

RedisClusterAsyncCommands<byte[], byte[]> getAsyncConnection() {
    //Redis事务才是 true
    if (this.isQueueing()) {
        return this.getAsyncDedicatedConnection();
    } else {
        //如果有共享连接则返回共享连接，否则返回独占连接，只有独占连接 PipeliningFlushPolicy 才会生效，PipeliningFlushPolicy 不会修改共享连接
        return (RedisClusterAsyncCommands)(this.asyncSharedConn != null && this.asyncSharedConn instanceof StatefulRedisConnection ? ((StatefulRedisConnection)this.asyncSharedConn).async() : this.getAsyncDedicatedConnection());
    }
}

由于我们要使用 PipeliningFlushPolicy，所以需要这里返回独占连接，也就不能打开共享连接。

我们来看下这个 PipeliningFlushPolicy 的源码就知道这个 flushOnClose 的意义：

public interface PipeliningFlushPolicy {
    //其实就是默认的每个命令都直接发到 Redis Server
    static PipeliningFlushPolicy flushEachCommand() {
    return FlushEachCommand.INSTANCE;
  }
  //在连接关闭的时候，将命令一起发到 Redis
  static PipeliningFlushPolicy flushOnClose() {
    return FlushOnClose.INSTANCE;
  }
  //手动设置在多少条命令之后，统一发到 Redis，但是同样的，连接关闭的时候也会发到 Redis
  static PipeliningFlushPolicy buffered(int bufferSize) {
    return () -> new BufferedFlushing(bufferSize);
  }
}

这三个类也都实现了 PipeliningFlushState 接口：

public interface PipeliningFlushState {
    //对于 executePipelined，刚开始就会调用 connection.openPipeline(); 开启 pipeline，里面会调用这个方法
    void onOpen(StatefulConnection<?, ?> connection);
    //对于 executePipelined 中的每个命令都会调用这个方法
    void onCommand(StatefulConnection<?, ?> connection);
    //在 executePipelined 的最后会调用 connection.closePipeline()，里面会调用这个方法
    void onClose(StatefulConnection<?, ?> connection);
}

默认的每个命令都直接发到 Redis Server 的实现是：其实就是方法里什么都不做。

private enum FlushEachCommand implements PipeliningFlushPolicy, PipeliningFlushState {
  INSTANCE;
  @Override
  public PipeliningFlushState newPipeline() {
    return INSTANCE;
  }
  @Override
  public void onOpen(StatefulConnection<?, ?> connection) {}
  @Override
  public void onCommand(StatefulConnection<?, ?> connection) {}
  @Override
  public void onClose(StatefulConnection<?, ?> connection) {}
}

对于 flushOnClose：

private enum FlushOnClose implements PipeliningFlushPolicy, PipeliningFlushState {
  INSTANCE;
  @Override
  public PipeliningFlushState newPipeline() {
    return INSTANCE;
  }
  @Override
  public void onOpen(StatefulConnection<?, ?> connection) {
      //首先配置连接的 AutoFlushCommands 为 false，这样命令就不会立刻发到 Redis
    connection.setAutoFlushCommands(false);
  }
  @Override
  public void onCommand(StatefulConnection<?, ?> connection) {
        //收到命令时什么都不做
  }
  @Override
  public void onClose(StatefulConnection<?, ?> connection) {
      //在 pipeline 关闭的时候发送所有命令
    connection.flushCommands();
    //恢复默认配置，这样连接如果退回连接池不会影响后续使用
    connection.setAutoFlushCommands(true);
  }
}

对于 buffered：

private static class BufferedFlushing implements PipeliningFlushState {
  private final AtomicLong commands = new AtomicLong();
  private final int flushAfter;
  public BufferedFlushing(int flushAfter) {
    this.flushAfter = flushAfter;
  }
  @Override
  public void onOpen(StatefulConnection<?, ?> connection) {
      //首先配置连接的 AutoFlushCommands 为 false，这样命令就不会立刻发到 Redis
    connection.setAutoFlushCommands(false);
  }
  @Override
  public void onCommand(StatefulConnection<?, ?> connection) {
      //如果命令达到指定个数，就发到 Redis
    if (commands.incrementAndGet() % flushAfter == 0) {
      connection.flushCommands();
    }
  }
  @Override
  public void onClose(StatefulConnection<?, ?> connection) {
      //在 pipeline 关闭的时候发送所有命令
    connection.flushCommands();
    //恢复默认配置，这样连接如果退回连接池不会影响后续使用
    connection.setAutoFlushCommands(true);
  }
}

Spring-data-redis + Lettuce 如何使用 Pipeline

Redis Pipeline 类比 Lettuce 中的 AutoFlushCommands

配置 Spring-data-redis + Lettuce 使用 Pipeline

热门文章

最新文章

相关课程

相关电子书

相关实验场景

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

直播

下载

镜像站

技术资料

Spring-data-redis + Lettuce 如何使用 Pipeline

Redis Pipeline 类比 Lettuce 中的 AutoFlushCommands

配置 Spring-data-redis + Lettuce 使用 Pipeline

热门文章

最新文章

相关课程

相关电子书

相关实验场景