记一次Redis超时

关键字：#spring-data-redis、#RedisTemplate、#Pipeline、#Lettuce

spring-data-redis:2.6.3

1 现象

时间轴（已脱敏）

day01 线上发现接口耗时不正常变高

day02 其他接口mget操作偶现超时，陆续发现其他Redis命令也偶尔出现超时（持续半个月）

day03 排查Redis无慢查询，连接数正常，确认为批量写缓存导致

day04 尝试去除问题缓存，Redis超时消失，服务多个接口耗时下降50%～60%

day05 改进配置，重新上线，缓存正常，接口耗时波动不大

2 错误

2.1 spring-data-redis虚假的pipeline

需求：高频批量刷缓存，每个string key单独设置随机过期时间，单次批量操作上限为500。

spring-data-redis的multiSet不支持同时设置过期时间，但是spring-data-redis支持pipeline。

问题代码鉴赏

    /**
     * 批量缓存
     * @param time base过期时间
     * @param random 随机过期时间范围 1表示不增加随机范围
     */
    private void msetWithRandomExpire(Map<String, String> kv, long time, int random) {
        RedisSerializer<String> stringSerializer = template.getStringSerializer();
        Random rand = new Random();
        template.executePipelined((RedisCallback<String>) connection -> {
            connection.openPipeline();
            kv.forEach((k, v) -> {
                long expireTime = time + rand.nextInt(random);
                connection.setEx(Objects.requireNonNull(stringSerializer.serialize(k)),
                        expireTime, Objects.requireNonNull(stringSerializer.serialize(v)));
            });
            connection.closePipeline();
            return null;
        });
    }

测试发现redis连接超时。

spring-data-redis采用的默认Redis客户端是Lettuce，Lettuce所有请求默认使用同一个共享连接的实例，只有当执行事务/pipeline命令时会新建一个私有连接。

执行单个Redis命令时，每收到一条命令，Lettuce就发送给Redis服务器，而pipeline需要将批量的命令缓存在内存，然后一次性发送给Redis服务器。

但是，查看LettuceConnection源码发现，Lettuce默认的pipeline刷入方式是FlushEachCommand，也就是每条命令都会产生一次发送行为。

使用pipeline的本意是避免多次发送带来的网络开销，所以spring-data-redis的pipeline是个伪批量操作，本质上和一条一条发送没有区别。

// org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory#pipeliningFlushPolicy
public class LettuceConnection extends AbstractRedisConnection {
    // ...
    private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand();
    // ...
}

2.2 Lettuce手动刷入的并发问题

spring-data-redis对Lettuce的封装存在缺陷，考虑使用原生的Lettuce客户端实现pipeline。

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

在Lettuce官网找到了用异步方式实现的pipeline代码，参考官网样例后写出的问题代码如下：

    public void msetWithRandomExpire(Map<String, String> kv, long time, int random) {
        RedisConnectionFactory connectionFactory = redisTemplate.getConnectionFactory();
        LettuceConnection connection = null;
        RedisClusterAsyncCommands<byte[], byte[]> commands = null;
        try {
            Random rand = new Random();
            connection = (LettuceConnection) RedisConnectionUtils.getConnection(connectionFactory);
            commands = connection.getNativeConnection();
            commands.setAutoFlushCommands(false);
            List<RedisFuture<?>> futures = new ArrayList<>();
            for (Map.Entry<String, String> entry : kv.entrySet()) {
                String k = entry.getKey();
                String v = entry.getValue();
                long expireTime = time + rand.nextInt(random);
                futures.add(commands.setex(k.getBytes(), expireTime, v.getBytes()));
            }
			// 批量flush命令
            commands.flushCommands();

            LettuceFutures.awaitAll(5, TimeUnit.SECONDS, futures.toArray(new RedisFuture[futures.size()]));
        } finally {
            // 恢复自动刷入
            if (commands != null) {
                commands.setAutoFlushCommands(true);
            }
            if (connection != null) {
                RedisConnectionUtils.releaseConnection(connection, connectionFactory);
            }
        }
    }

官方声称这样写，在50-1000个批量操作的区间内，吞吐量可以提高五倍，简直完美满足我的需求。

上线测试后确实非常快，500个SETEX命令可以在10ms内完成，没有再发生过Redis连接超时的现象。

问题在于，AutoFlushCommands这个配置对于共享连接是全局的，会影响到其他正在使用共享连接的线程。

所以，Lettuce官方的建议是把这个操作放在一个私有连接里进行，这样就不会影响到共享连接中的命令。

The AutoFlushCommands state is set per connection and therefore affects all threads using the shared connection. If you want to omit this effect, use dedicated connections. The AutoFlushCommands state cannot be set on pooled connections by the Lettuce connection pooling.

spring-data-redis里执行pipeline命令，会先申请一个私有连接，虽然它的刷入命令的策略有问题，但这个可以参考下。

翻了下Lettuce的API，发现通过getNativeConnection方法可以获取到私有连接。

connection = (LettuceConnection) RedisConnectionUtils.getConnection(connectionFactory);
commands = connection.getNativeConnection();

@Override
public RedisClusterAsyncCommands<byte[], byte[]> getNativeConnection() {
		LettuceSubscription subscription = this.subscription;
		// getAsyncConnection()会返回一个私有连接
		return (subscription != null ? subscription.getNativeConnection().async() : getAsyncConnection());
}

研究到这里以为大功告成，由于用了比较取巧的写法，上线也观察了两天，并没有出现问题，直到第三天排行榜Redis莫名开始出现超时。

报错如下

io.lettuce.core.RedisCommandTimeoutException: Command timed out after 1 minute(s)

在Github issue翻到一个老哥遇到了同样的问题，Lettuce作者的回答是

Switching setAutoFlushCommands to false is only recommended for single-threaded connection use that wants to optimize command buffering for batch imports.

Lettuce works in general in a non-blocking, multiplexing mode regardless of the API that you're using. You can use synchronous, asynchronous, any reactive APIs with the same connection.

That being said, if you don't touch setAutoFlushCommands, you should be good.

只推荐在单线程的应用中使用setAutoFlushCommands来手动刷命令。Lettuce通常以非阻塞、多路复用模式工作，与使用什么API无关，不管是同步/异步/响应式API。如果你不碰这东西，就没事了。

作者跟官网说的有点矛盾，官网强调了只要在私有连接里进行pipeline操作就不会影响到共享连接，所以怀疑到底有没有正确获取到私有连接。

回到Lettuce的API，getNativeConnection这个方法再点进去一层

	RedisClusterAsyncCommands<byte[], byte[]> getAsyncConnection() {

		if (isQueueing() || isPipelined()) {
			return getAsyncDedicatedConnection();
		}
        // 当共享连接不为空 返回一个共享连接
		if (asyncSharedConn != null) {

			if (asyncSharedConn instanceof StatefulRedisConnection) {
				return ((StatefulRedisConnection<byte[], byte[]>) asyncSharedConn).async();
			}
			if (asyncSharedConn instanceof StatefulRedisClusterConnection) {
				return ((StatefulRedisClusterConnection<byte[], byte[]>) asyncSharedConn).async();
			}
		}
		return getAsyncDedicatedConnection();
	}

原来getNativeConnection这个方法获取私有连接是有条件的，只有当共享连接被关闭时才会返回私有连接。

而关闭共享连接需要调用setShareNativeConnection(false)这个方法，这个配置同样是全局的，关闭后，所有的命令都会走私有连接，这时需要用连接池来管理Lettuce连接。

到这里Redis超时的原因就找到了。

Lettuce官方在文档最后的QA里贴了一个出现RedisCommandTimeoutException的可能原因，最后一条是：【为什么要贴在最后…】

If you manually control the flushing behavior of commands (setAutoFlushCommands(true/false)), you should have a good reason to do so. In multi-threaded environments, race conditions may easily happen, and commands are not flushed. Updating a missing or misplaced flushCommands() call might solve the problem.

意思是，修改在AutoFlushCommands这个配置的时候需要注意，多线程环境中，竞态会频繁出现，命令将会阻塞，修改在不当的场景下使用手动刷入flushCommands也许会解决问题。

【以下为个人理解】

虽然在finally中恢复了自动刷入，但是在并发场景下，会有一些在AutoFlushCommands=false时执行的命令，这些命令将会被阻塞在本地内存，无法发送到Redis服务器。所以这个问题本质是网络的阻塞，通过info clients查询Redis连接数正常，配置超时没有用，慢日志也查不到任何记录，干掉缓存的批量操作后，Redis终于正常了。

3 修复

在上面的前提下修复这个问题，需要三步

3.1 配置

3.1.1 Lettuce连接池

所有命令走单独的私有连接，需要用连接池管理。

具体参数根据业务调整

spring.redis.lettuce.pool.max-active=50  
# Minimum number of idle connections in the connection pool.
spring.redis.lettuce.pool.min-idle=5  
# Maximum number of idle connections in the connection pool.
spring.redis.lettuce.pool.max-idle=50  
# Maximum time for waiting for connections in the connection pool. A negative value indicates no limit.
spring.redis.lettuce.pool.max-wait=5000  
# Interval for scheduling an eviction thread.
spring.redis.pool.time-between-eviction-runs-millis=2000

3.1.2 关闭共享连接

搭配spring-data-redis使用，关闭共享连接

@Bean
public LettuceConnectionFactory lettuceConnectionFactory() {
    LettuceConnectionFactory factory = new LettuceConnectionFactory();
    factory.setShareNativeConnection(true);
    // read config
    return factory;
}

3.2 写法调整

3.2.1 用Lettuce API重写代码

    public void msetWithRandomExpire(Map<String, String> kv, long baseTime, int random) {
        RedisClient client = RedisClient.create();
        try (StatefulRedisConnection<String, String> connection = client.connect()) {
            Random rand = new Random();
            RedisAsyncCommands<String, String> commands = connection.async();
            // 关闭命令自动flush
            commands.setAutoFlushCommands(false);
            List<RedisFuture<?>> futures = new ArrayList<>();
            for (Map.Entry<String, String> entry : kv.entrySet()) {
                long expireTime = baseTime + rand.nextInt(random);
                futures.add(commands.setex(entry.getKey(), expireTime, entry.getValue()));
            }
            // 手动批量flush
            commands.flushCommands();
            LettuceFutures.awaitAll(5, TimeUnit.SECONDS, futures.toArray(new RedisFuture[0]));
        }
    }

3.2.2 spring-data-redis封装的flush策略

除了用Lettuce原生API实现之外，spring-data-redis也已经给pipeline封装好了三种flush策略。

PipeliningFlushPolicy也就是Lettuce的pipeline刷新策略，包括默认的每个命令都刷入，一共有三种，基本上满足大部分业务场景。

  /** 
    * org.springframework.data.redis.connection.lettuce.LettuceConnection.PipeliningFlushPolicy
    * FlushEachCommand: 每个命令flush一次 默认策略
    * FlushOnClose: 每次连接关闭时flush一次
    * BufferedFlushing: 设置buffer大小 每达到buffer个命令刷一次 连接关闭时也刷一次
    */
	public interface PipeliningFlushPolicy {

		static PipeliningFlushPolicy flushEachCommand() {
			return FlushEachCommand.INSTANCE;
		}
    
		static PipeliningFlushPolicy flushOnClose() {
			return FlushOnClose.INSTANCE;
		}
    
		static PipeliningFlushPolicy buffered(int bufferSize) {

			Assert.isTrue(bufferSize > 0, "Buffer size must be greater than 0");
			return () -> new BufferedFlushing(bufferSize);
		}

		PipeliningFlushState newPipeline();
	}

设置pipeliningFlushPolicy=FlushOnClose之后，上面在2.1节提到的虚假的pipeline就成为真正的pipeline了。

@Bean
public LettuceConnectionFactory lettuceConnectionFactory() {
    LettuceConnectionFactory factory = new LettuceConnectionFactory();
    factory.setShareNativeConnection(true);
    // 设置pipeline的flush策略
    factory.setPipeliningFlushPolicy(LettuceConnection.PipeliningFlushPolicy.flushOnClose());
    // read config
    return factory;
}