[大数据]如何正确使用Vertx操作Redis（3.9.4带源码分析和测试结果）

1 缘起

真的是一段特别的经历。
增加了知识储备：Vert.x异步连接Redis并操作Redis。
事情经过：
团队需要开发离线算法中间件，集成到业务侧后台服务中，
原有的中间件中使用哨兵方式操作Redis，现在需要增加新的逻辑，
使用集群方式操作Redis，中间件中基于Vert.x 3.5.1修改出了一个支持集群连接的版本，
满足集群连接和操作。一切都很美好，测试通过。
于是，自己测试原生Vertx高版本（>3.8.x）的集群连接，这才是噩梦的开始，
由于我自建的Redis集群有密码认证，
所以，使用了Vert.x不同版本测试都无法连接，出于无奈，只能，去掉Redis集群的密码认证，
但是，3.8.x仍旧无法连接，
最后，使用了3.9.4，连接成功，并成功完成Redis操作，
特分享如下，帮助未来有需要的开发者。

特别声明：

***Vert.x版本：3.9.4
本文实测：单体Redis、哨兵和集群Redis，并给出测试结果；

Redis环境搭建参考：
集群搭建：Ubuntu20部署Redis6.0(伪)集群
哨兵搭建：CentOS7部署哨兵Redis（带架构图，清晰易懂）

2 Vert.x依赖

<!-- https://mvnrepository.com/artifact/io.vertx/vertx-core -->
<dependency>
    <groupId>io.vertx</groupId>
    <artifactId>vertx-core</artifactId>
    <version>3.9.4</version>
</dependency>

<!-- https://mvnrepository.com/artifact/io.vertx/vertx-redis-client -->
<dependency>
    <groupId>io.vertx</groupId>
    <artifactId>vertx-redis-client</artifactId>
    <version>3.9.4</version>
</dependency>

3 Redis地址库

为统一管理Redis地址，使用枚举管理，
后续添加Redis连接配置时，从这个枚举库中提取，
代码如下所示：

package com.monkey.java_study.common.enums;

/**
 * Redis地址库.
 *
 * @author xindaqi
 * @since 2022-07-01 11:10
 */
public enum RedisAddressEnum {

    // Redis集群地址
    CLUSTER_NODE_1("redis://192.168.211.129:9001"),
    CLUSTER_NODE_2("redis://192.168.211.129:9002"),
    CLUSTER_NODE_3("redis://192.168.211.129:9003"),
    CLUSTER_NODE_4("redis://192.168.211.129:9004"),
    CLUSTER_NODE_5("redis://192.168.211.129:9005"),
    CLUSTER_NODE_6("redis://192.168.211.129:9006"),

    // Redis哨兵地址
    SENTINEL_1("redis://192.168.1.12:26379"),
    SENTINEL_2("redis://192.168.1.12:26380"),
    SENTINEL_3("redis://192.168.1.12:26381"),

    // Redis单体地址
    STANDALONE_NODE("redis://:123456@127.0.0.1/0"),
    ;

    private String address;

    RedisAddressEnum(String address) {
        this.address = address;
    }

    public String getAddress() {
        return address;
    }
}

4 Redis集群

集群搭建：Ubuntu20部署Redis6.0(伪)集群
哨兵搭建：CentOS7部署哨兵Redis（带架构图，清晰易懂）

4.1 配置地址和相关参数：RedisOptions

Vert.x 3.9.4提供了两种方式配置Redis地址，
分别为：addConnectionString和setEndpoints，
同时，可以指定连接类型，集群为：RedisClientType.CLUSTER
特别注意：集群方式连接时，集群千万不可配置密码认证，否则无法连接到Redis集群

• addConnectionString方式
  此种方式，地址一条一条地添加。地址格式：
  redis://ip:port

/**
     * Redis集群连接配置：addConnectionString方式
     *
     * @return Redis集群配置
     */
    private static RedisOptions setClusterRedisAddressOneByOne() {
        return new RedisOptions()
                .setType(RedisClientType.CLUSTER)
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_1.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_2.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_3.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_4.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_5.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_6.getAddress());
    }

• setEndpoints方式
  批量添加Redis地址，endpoints的类型为List<String>，因此可以批量添加。

    /**
     * Redis集群连接配置：setEndpoints方式
     *
     * @return Redis集群配置
     */
    private static RedisOptions setClusterRedisAddressBatch() {
        List<String> redisAddressList = Stream.of(
                RedisAddressEnum.CLUSTER_NODE_1.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_2.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_3.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_4.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_5.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_6.getAddress()).collect(Collectors.toList());
        return new RedisOptions()
                .setType(RedisClientType.CLUSTER)
                .setEndpoints(redisAddressList);
    }

4.2 源码：连接类型

由Vert.x3.9.4源码可知，Redis客户端连接类型有三种：STANDALONE（单体）、SENTINEL（哨兵）和CLUSTER（集群），
源码如下图所示。
位置：io.vertx.redis.client.RedisClientType

4.3 其他默认参数

配置Redis集群连接地址后，就可以直接连接Redis集群，
因为，Vert.x3.9.4在初始化时，自动配置了默认值，如：
maxWaitingHandler、maxPoolSize和maxPoolWaiting等，源码如下图所示，
默认情况，连接Redis的类型为单体（RedisClientType.STANDALONE），
哨兵模式下的默认不使用从节点：RedisSlaves.NEVER。
位置：io.vertx.redis.client.RedisOptions#init

为什么会加载这些参数，因为，RedisOptions的构造函数：
位置：io.vertx.redis.client.RedisOptions#RedisOptions()

4.4 创建Redis客户端

创建样例如下，通过传入Vertx.vertx()和redisOptions（上文配置的参数对象），
即可获取Redis客户端，获取客户端即与Redis建立起连接，
接下来即可完成CURD操作。

Redis.createClient(Vertx.vertx(), redisOptions);

4.5 源码：Redis.createClient

Vert.x 3.9.4创建Redis客户端源码如下图所示，
由源码可知，创建过程中会根据type类型选择不同的客户端（单体、哨兵和集群）。
位置：io.vertx.redis.client.Redis#createClient(io.vertx.core.Vertx, io.vertx.redis.client.RedisOptions)

4.6 Redis操作：RedisAPI

Vert.x 3.9.4通过RedisAPI操作Redis数据，
测试样例如下，使用Vert.x可知，异步操作，
因此，读取数据使用Vetx.x线程处理：vert.x-eventloop-thread-0

/**
     * 读取数据从Redis
     *
     * @param redisClient Redis客户端
     * @param key         Redis中查询的键
     */
    public void readDataFromCluster(Redis redisClient, String key) {
        RedisAPI redisAPI = RedisAPI.api(redisClient);
        redisAPI.get(key, res -> {
            if (res.succeeded()) {
                String value = String.valueOf(res.result());
                logger.info(">>>>>>>>Read data from redis cluster (key, value)->:({},{})", key, value);
            } else {
                logger.error(">>>>>>>>Redis cluster read data error:", res.cause());
            }
        });
    }

4.7 完整样例

package com.monkey.java_study.thirdparty.vertx_test;

import com.monkey.java_study.common.enums.RedisAddressEnum;
import io.vertx.core.Vertx;
import io.vertx.redis.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Vert.x连接Redis集群服务和读数据测试.
 *
 * @author xindaqi
 * @since 2022-06-30 12:16
 */
public class VertxClusterOpRedisTest {

    private static final Logger logger = LoggerFactory.getLogger(VertxClusterOpRedisTest.class);

    /**
     * Redis集群客户端
     */
    private static final Redis redisClient;

    /**
     * Redis连接配置
     */
    private static final RedisOptions redisOptions;

    static {
        redisOptions = setClusterRedisAddressBatch();
        redisClient = Redis.createClient(Vertx.vertx(), redisOptions);
    }

    /**
     * Redis集群连接配置：addConnectionString方式
     *
     * @return Redis集群配置
     */
    private static RedisOptions setClusterRedisAddressOneByOne() {
        return new RedisOptions()
                .setType(RedisClientType.CLUSTER)
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_1.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_2.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_3.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_4.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_5.getAddress())
                .addConnectionString(RedisAddressEnum.CLUSTER_NODE_6.getAddress());
    }

    /**
     * Redis集群连接配置：setEndpoints方式
     *
     * @return Redis集群配置
     */
    private static RedisOptions setClusterRedisAddressBatch() {
        List<String> redisAddressList = Stream.of(
                RedisAddressEnum.CLUSTER_NODE_1.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_2.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_3.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_4.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_5.getAddress(),
                RedisAddressEnum.CLUSTER_NODE_6.getAddress()).collect(Collectors.toList());
        return new RedisOptions()
                .setType(RedisClientType.CLUSTER)
                .setEndpoints(redisAddressList);
    }

    public Redis getRedisClient() {
        return redisClient;
    }

    /**
     * 读取数据从Redis
     *
     * @param redisClient Redis客户端
     * @param key         Redis中查询的键
     */
    public void readDataFromCluster(Redis redisClient, String key) {
        RedisAPI redisAPI = RedisAPI.api(redisClient);
        redisAPI.get(key, res -> {
            if (res.succeeded()) {
                String value = String.valueOf(res.result());
                logger.info(">>>>>>>>Read data from redis cluster (key, value)->:({},{})", key, value);
            } else {
                logger.error(">>>>>>>>Redis cluster read data error:", res.cause());
            }
        });
    }

    public static void main(String[] args) {
        VertxClusterOpRedisTest vertxOpRedisTest = new VertxClusterOpRedisTest();
        vertxOpRedisTest.readDataFromCluster(vertxOpRedisTest.getRedisClient(), "name");
    }
}

4.8 测试结果

由测试结果可知，Vert.x 3.9.4操作Redis使用异步线程：vert.x-enventloop-thread-0，
并完成结果读取。

5 Redis单体

有了上面集群配置的基础，
配置单体连接和哨兵连接就相对好理解一些，
直接给出完整样例，代码如下：
需要注意的是：配置Redis时需要按照需要配置数据库0-15，
连接类型选择：RedisClientType.STANDALONE，
如果是单体可以不选，因为RedisOptions默认值即为：RedisClientType.STANDALONE。
地址格式：
redis://username:password@ip:port/db

参数说明：

特别地：
如果只有密码没有用户名，应该这样配置：redis://:password@ip:port/db

5.1 完整样例

package com.monkey.java_study.thirdparty.vertx_test;

import com.monkey.java_study.common.enums.RedisAddressEnum;
import io.vertx.core.Vertx;
import io.vertx.redis.client.Redis;
import io.vertx.redis.client.RedisAPI;
import io.vertx.redis.client.RedisClientType;
import io.vertx.redis.client.RedisOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Vert.x连接Redis单体服务和读数据测试.
 *
 * @author xindaqi
 * @since 2022-06-30 12:16
 */
public class VertxStandaloneOpRedisTest {

    private static final Logger logger = LoggerFactory.getLogger(VertxStandaloneOpRedisTest.class);

    /**
     * Redis单体客户端
     */
    private static final Redis redisClient;

    /**
     * Redis连接配置
     */
    private static final RedisOptions redisOptions;

    static {
        redisOptions = setStandaloneRedisAddress();
        redisClient = Redis.createClient(Vertx.vertx(), redisOptions);
    }

    /**
     * 配置Redis单体连接地址
     *
     * @return Redis单体配置
     */
    private static RedisOptions setStandaloneRedisAddress() {
        return new RedisOptions()
                .setType(RedisClientType.STANDALONE)
                .addConnectionString(RedisAddressEnum.STANDALONE_NODE.getAddress());
    }


    public Redis getRedisClient() {
        return redisClient;
    }

    /**
     * 读取数据从Redis
     *
     * @param redisClient Redis客户端
     * @param key         Redis中查询的键
     */
    public void readDataFromStandalone(Redis redisClient, String key) {
        RedisAPI redisAPI = RedisAPI.api(redisClient);
        redisAPI.get(key, res -> {
            if (res.succeeded()) {
                String value = String.valueOf(res.result());
                logger.info(">>>>>>>>Read data from redis standalone (key, value)->:({},{})", key, value);
            } else {
                logger.error(">>>>>>>>Redis standalone read data error:", res.cause());
            }
        });
    }


    public static void main(String[] args) {
        VertxStandaloneOpRedisTest vertxOpRedisTest = new VertxStandaloneOpRedisTest();
        vertxOpRedisTest.readDataFromStandalone(vertxOpRedisTest.getRedisClient(), "name");
    }
}

5.2 测试结果

6 哨兵模式

集群搭建：Ubuntu20部署Redis6.0(伪)集群
哨兵搭建：CentOS7部署哨兵Redis（带架构图，清晰易懂）
同理，完整样例如下，
需要注意的是，类型配置为：RedisClientType.SENTINEL，
同时需要配置主服务的名称和角色，即
MasterName(“mymaster”)和Role(RedisRole.MASTER)，
其中，哨兵模式下连接哨兵，而不是直接连接Redis集群，
即地址为哨兵的IP和Port。

package com.monkey.java_study.thirdparty.vertx_test;

import com.monkey.java_study.common.enums.RedisAddressEnum;
import io.vertx.core.Vertx;
import io.vertx.redis.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Vert.x连接Redis哨兵服务和读数据测试.
 *
 * @author xindaqi
 * @since 2022-06-30 12:16
 */
public class VertxSentinelOpRedisTest {

    private static final Logger logger = LoggerFactory.getLogger(VertxSentinelOpRedisTest.class);

    /**
     * Redis哨兵客户端
     */
    private static final Redis redisClient;

    /**
     * Redis连接配置
     */
    private static final RedisOptions redisOptions;

    static {
        redisOptions = setSentinelRedisAddressOneByOne();
        redisClient = Redis.createClient(Vertx.vertx(), redisOptions);
    }

    /**
     * Redis哨兵连接配置：addConnectionString方式
     *
     * @return Redis集群配置
     */
    private static RedisOptions setSentinelRedisAddressOneByOne() {
        return new RedisOptions()
                .setType(RedisClientType.SENTINEL)
                .addConnectionString(RedisAddressEnum.SENTINEL_1.getAddress())
                .addConnectionString(RedisAddressEnum.SENTINEL_2.getAddress())
                .addConnectionString(RedisAddressEnum.SENTINEL_3.getAddress())
                .setMasterName("mymaster")
                .setRole(RedisRole.MASTER);
    }

    /**
     * Redis哨兵连接配置：setEndpoints方式
     *
     * @return Redis集群配置
     */
    private static RedisOptions setSentinelRedisAddressBatch() {
        List<String> redisAddressList = Stream.of(
                RedisAddressEnum.SENTINEL_1.getAddress(),
                RedisAddressEnum.SENTINEL_2.getAddress(),
                RedisAddressEnum.SENTINEL_3.getAddress()).collect(Collectors.toList());
        return new RedisOptions()
                .setType(RedisClientType.SENTINEL)
                .setMasterName("mymaster")
                .setRole(RedisRole.MASTER)
                .setEndpoints(redisAddressList);
    }

    public Redis getRedisClient() {
        return redisClient;
    }

    /**
     * 读取数据从Redis
     *
     * @param redisClient Redis客户端
     * @param key         Redis中查询的键
     */
    public void readDataFromSentinel(Redis redisClient, String key) {
        RedisAPI redisAPI = RedisAPI.api(redisClient);
        redisAPI.get(key, res -> {
            if (res.succeeded()) {
                String value = String.valueOf(res.result());
                logger.info(">>>>>>>>Read data from redis sentinel (key, value)->:({},{})", key, value);
            } else {
                logger.error(">>>>>>>>Redis sentinel read data error:", res.cause());
            }
        });
    }

    public static void main(String[] args) {
        VertxSentinelOpRedisTest vertxOpRedisTest = new VertxSentinelOpRedisTest();
        vertxOpRedisTest.readDataFromSentinel(vertxOpRedisTest.getRedisClient(), "name");
    }
}

测试结果

7 小结

Vert.x 3.9.4连接和操作Redis核心：
（1）配置Redis连接地址，通过RedisOptions，该类实例化时，有默认的参数；
（2）Vert.x3.9.4提供的Redis客户端无法连接需要密码认证的Redis集群，但是，可以连接单体需要密码认证的Redis；
（3）创建连接使用Redis.createClient，该方法根据type选择建立对应的Redis客户端：集群（cluster）、哨兵（sentinel）和单体（standalone），其中哨兵方式直接连接哨兵地址，由哨兵分配Redis主集群读写；
（4）操作Redis使用RedisAPI完成；
（5）Vert.x操作Redis是通过异步线程完成：Vert.x的线程，如vert.x-eventloop-thread-0。