验证Lettuce在单连接上进行多路复用-Toy模板网

这篇具有很好参考价值的文章主要介绍了验证Lettuce在单连接上进行多路复用。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

redis multiplexing机制类似http2，在存在并发请求时能减少大量通讯延迟，但不支持blocking相关的操作(如BLPOP)

配置 RedisTemplate

RedisTemplate自动根据操作类型，选择是在单连接上进行多路复用(setex/get)，还是申请新的连接/等待空闲连接(blpop)

@Configuration
public class RedisTemplateConfig {
   @Bean
   public LettuceConnectionFactory lettuceConnectionFactory(@Value("${redis.cluster}") String address,
           @Value("${redis.password}") String password) {
       // 配置连接池管理
       var poolConfig = new GenericObjectPoolConfig<StatefulRedisClusterConnection<String, String>>();
       poolConfig.setMaxTotal(20);
       poolConfig.setMaxIdle(20);
       poolConfig.setMinIdle(2);
       poolConfig.setTestWhileIdle(true);
       poolConfig.setMinEvictableIdleDuration(Duration.ofMillis(60000));
       poolConfig.setTimeBetweenEvictionRuns(Duration.ofMillis(30000));
       poolConfig.setNumTestsPerEvictionRun(-1);
       // 配置客户端行为
       var clientConfig = LettucePoolingClientConfiguration.builder()
               .clientOptions(ClusterClientOptions.builder()
                       .autoReconnect(true)
                       .pingBeforeActivateConnection(true)
                       .socketOptions(SocketOptions.builder().connectTimeout(Duration.ofSeconds(3)).build())
                       .timeoutOptions(TimeoutOptions.builder().fixedTimeout(Duration.ofSeconds(3)).build())
                       .topologyRefreshOptions(ClusterTopologyRefreshOptions.builder()
                               .enableAdaptiveRefreshTrigger(RefreshTrigger.MOVED_REDIRECT,
                                       RefreshTrigger.PERSISTENT_RECONNECTS)
                               .adaptiveRefreshTriggersTimeout(Duration.ofSeconds(30))
                               .build())
                       .build())
               .poolConfig(poolConfig)
               .build();
       // 配置集群连接信息
       var redisConfig = new RedisClusterConfiguration();
       redisConfig.setMaxRedirects(5);
       redisConfig.setPassword(password);
       String[] serverArray = address.split(",|，|;|；");// 获取服务器数组
       Set<RedisNode> nodes = new HashSet<>();
       for (String ipPort : serverArray) {
           nodes.add(RedisNode.fromString(ipPort));
       }
       redisConfig.setClusterNodes(nodes);
       return new LettuceConnectionFactory(redisConfig, clientConfig);
   }

   @Bean
   public StringRedisTemplate redisTemplate(LettuceConnectionFactory lettuceConnectionFactory) {
       return new StringRedisTemplate(lettuceConnectionFactory);
   }
}

验证blocking操作在独立连接上进行

实现blocking LPOP操作的方法在opsForList()里；
并发100压测期间查看客户端本地的tcp连接，可以看到和每个redis节点都建立了大量连接；
证明RedisTemplate没有选择让不同并发线程共用同一个StatefulConnection

    @Test
    public void blpop() throws Exception {
        long start = System.currentTimeMillis();
        AtomicLong err = new AtomicLong();
        int maxThreads = 100;
        Semaphore semaphore = new Semaphore(maxThreads);
        for (int i = 0; i < maxThreads; i++) {
            final int threadnum = i + 1;
            semaphore.acquire(1);
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 这里是blocking LPOP
                        template.opsForList().leftPop("test" + threadnum, 2, TimeUnit.SECONDS);
                    } catch (Exception ex) {
                        log.error("leftPop", ex);
                        err.addAndGet(1L);
                    } finally {
                        semaphore.release(1);
                    }
                }
            }).start();
        }
        semaphore.acquire(maxThreads);
        long end = System.currentTimeMillis();
        log.info("耗时{}ms, 错误{}", end - start, err.get());
    }

Lettuce在这个场景下使用一个NIO selector/multiplexer来收发多个tcp连接上的消息，但是在单个tcp连接上没有使用全双工通讯和multiplexing（多路复用）；
Lettuce的StatefulConnection是线程安全的，如果多线程共用一个StatefulConnection来执行blocking LPOP操作，那么会串行执行，严重影响执行效率（可通过操作Lettuce原生接口进行验证）

验证单tcp连接的多路复用

发起100个线程，每个线程连续进行1000次setex + get操作；
执行期间查看客户端本地的tcp连接，可以看到只建立了一个和redis节点的连接；
redis通讯协议在tcp全双工的基础上实现了多路复用，支持多个指令随时并发地在同一个tcp连接上发出和响应，且每个指令的发出和响应互不干扰，与http2的各路stream互不干扰类似；
如果和使用半双工通讯的Jedis进行对比，需要配置Jedis连接池maxTotal和maxIdle连接数都到100，才能有相同通讯效率

@Slf4j
@EnableAutoConfiguration(exclude = { RedisAutoConfiguration.class })
@SpringBootTest(classes = { RedisTemplateConfig.class })
public class RedisTemplateTest {
    @Autowired
    StringRedisTemplate template;

    @Test
    public void getSet() throws Exception {
        long start = System.currentTimeMillis();
        int maxThreads = 100;
        long maxMessagess = 1000;
        AtomicLong err = new AtomicLong();
        AtomicLong num = new AtomicLong();
        Semaphore semaphore = new Semaphore(maxThreads);
        for (int i = 0; i < maxThreads; i++) {
            final int threadnum = i + 1;
            semaphore.acquire(1);
            new Thread(new Runnable() {
                @Override
                public void run() {
                    int j = 0;
                    try {
                        for (; j < maxMessagess; j++) {
                            String key = "thread" + threadnum + "test" + j;
                            String value = "test" + j;
                            template.opsForValue().set(key, value, 1, TimeUnit.SECONDS);
                            assertEquals(value, template.opsForValue().get(key));
                        }
                    } finally {
                        num.addAndGet(j);
                        semaphore.release(1);
                    }
                }
            }).start();
        }
        semaphore.acquire(maxThreads);
        long end = System.currentTimeMillis();
        double rate = 1000d * num.get() / (end - start);
        log.info("每秒发送并读取消息{}; 耗时{}ms, 累计发送{}", rate, end - start, num.get());
    }