@[toc]

前言

bala bala...

vertx是什么

vert.x是一个全异步框架，通过内部的event bus跟event loop做到了处处皆异步。关于里面的细节，后面我会出一篇文章详细跟大家探讨...

为什么要用vertx

充分利用机器性能，提升应用的吞吐量上限。

vertx-redis-client是什么

通俗来讲，vertx-redis提供了一个全异步的可配置的redis客户端

vertx-redis-client有哪些特性

相对常规的jedis或者Lettuce来说，这里的特性其实就是vertx贯穿全局的特性：异步。当我们调用api发送一条redis指令的时候，可以不用等待redis的响应，只需要绑定一个回调函数，程序就可以继续往下执行。当拿到redis的响应之后，就会触发这个回调。当然，像这种编程方式，需要把业务依赖关系处理得很明确，因为我们必须把依赖这个redis响应的所有处理都移到这个回调函数里去。

怎么使用vertx-redis-client

添加vertx-redis-client依赖

//gradle
compile("io.vertx:vertx-redis-client:$version")
//maven
<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-redis-client</artifactId>
  <vertion></version>
</dependency>

RedisConnection

RedisConnection是vertx-redis-client里暴露出来可操作redis的最基本的一组api了，使用如下

  Redis
    .createClient(Vertx.vertx(), "redis://ip:port")
    .connect(onConnect -> {
       //连接成功
       if (onConnect.succeeded()) {
           //获取连接实例
           RedisConnection conn = onConnect.result();
           //创建redis get指令. 
           Request command = Request.cmd(Command.GET);
           //添加指令参数aa. 最后到达redis的指令就是get aa
           command.arg("aa");
           //发送指令
           conn.send(command, resp -> {
               //执行成功
               if (resp.succeeded()) {
                   //拿到指令执行结果
                   Response result = resp.result();
               }
           });
       }
   });

这里就谈谈怎么使用吧，其实内部的实现就是通过netty包装的channel跟redis建立起连接，有兴趣的朋友可以点进去看看。

redisOptions

在介绍该redisClient各种模式之前，我们需要了解options是什么。
简单来说，redisOptions是我们用来定制该redisClient的一种方式，像之前我们通过==Redis.createClient(Vertx.vertx(), "redis://ip:port")== 这种仅仅提供一个地址的方式就建立起了一个redisClient，但是如果我们想要定制更复杂的redisClient呢？这个时候就需要用到redisOptions了

各项主要配置的含义

先大概看下redisOptions都有哪些属性：

public class RedisOptions {

  /**
   * The default redis endpoint = {@code redis://localhost:6379}
   */
  public static final String DEFAULT_ENDPOINT = "redis://localhost:6379";
  
  private RedisClientType type;
  private NetClientOptions netClientOptions;
  private List<String> endpoints;
  private int maxWaitingHandlers;
  private int maxNestedArrays;
  private String masterName;
  private RedisRole role;
  private RedisSlaves slaves;
  private String password;

  // pool related options
  private int poolCleanerInterval;
  private int maxPoolSize;
  private int maxPoolWaiting;
  private int poolRecycleTimeout;
}

type

该枚举用于指定redisClient以什么模式去连接server，这里的模式跟redis server的模式一一对应，我们在用的时候也必须注意，server是以什么模式部署的，client就以什么模式去配置。源码如下

public enum RedisClientType {

  /**
   * 默认值，单机模式
   */
  STANDALONE,

  /**
   * 哨兵
   */
  SENTINEL,

  /**
   * 集群
   */
  CLUSTER
}

这些配置决定了redisClient与server交互的方式，不同的类型会用不同的策略来应对，在我们调用Redis#creatClient的时候就会作出处理，源码如下

static Redis createClient(Vertx vertx, RedisOptions options) {
    switch (options.getType()) {
      case STANDALONE:
        return new RedisClient(vertx, options);
      case SENTINEL:
        return new RedisSentinelClient(vertx, options);
      case CLUSTER:
        return new RedisClusterClient(vertx, options);
      default:
        throw new IllegalStateException("Unknown Redis Client type: " + options.getType());
    }
  }

至于不同Client内部是怎么处理的，后面会一一介绍

masterName

这个只在Sentinel模式下有用，跟redisServer的redis_sentinel.conf里配置的masterName是一个东西，用来指明连接的是被哨兵监控的哪一个主从集群

role

这项配置只用于clientType为Sentinel的情况，主要用来指定当前的客户端到底是跟master连接还是只跟slave连接，如果配置为slave，则通过该客户端发起的所有指令都会被路由到slave节点，这意味着这个client对redisServer的所有操作都是只读的。
还有一个sentinel，这个意味着这个client只操作server端的sentinel，不会主动去获取到master或者slave的连接。按我目前看来，这个枚举值其实是给程序内部使用的，除非我们想手动实现跟哨兵的通信以及维护相应主从节点的一个failover情况，那么就可以使用这个配置。核心源码在RedisSentinelClient中，如下所示

private void createConnectionInternal(RedisOptions options, RedisRole role, Handler<AsyncResult<RedisConnection>> onCreate) {
    switch (role) {
      case SENTINEL:
        resolveClient(this::isSentinelOk, options, createAndConnect);
        break;

      case MASTER:
        resolveClient(this::getMasterFromEndpoint, options, createAndConnect);
        break;

      case SLAVE:
        resolveClient(this::getSlaveFromEndpoint, options, createAndConnect);
    }
  }

这里可以看到，根据不同的role，获取了不同类型的redis连接地址

private void getMasterFromEndpoint(String endpoint, RedisOptions options, Handler<AsyncResult<String>> handler) {
  final RedisURI uri = new RedisURI(endpoint);
  connectionManager.getConnection(context, getSentinelEndpoint(uri), null, onCreate -> {
    final RedisConnection conn = onCreate.result();
    final String masterName = options.getMasterName();
    
    // 根据masterName从sentinel获取相应主从集群的master节点信息
    conn.send(cmd(SENTINEL).arg("GET-MASTER-ADDR-BY-NAME").arg(masterName), getMasterAddrByName -> {
      //bala bala...
    });
  });

private void getSlaveFromEndpoint(String endpoint, RedisOptions options, Handler<AsyncResult<String>> handler) {
  final RedisURI uri = new RedisURI(endpoint);
    connectionManager.getConnection(context, getSentinelEndpoint(uri), null, onCreate -> {
      
      final RedisConnection conn = onCreate.result();
      final String masterName = options.getMasterName();
      // 获取masterName指定主从的slave节点
      conn.send(cmd(SENTINEL).arg("SLAVES").arg(masterName), sentinelSlaves -> {
        //...
      });
    });
  }

slaves

这项配置用来指定在集群模式下对server的读操作的行为，源码如下

public enum RedisSlaves {

  /**
   * 读操作只落在master
   */
  NEVER,

  /**
   * 读操作会随机落在master跟slave
   */
  SHARE,

  /**
   * 读操作只落在slave上
   */
  ALWAYS
}

其核心原理是通过对slave节点执行readonly命令来开启slave的查询功能，因为默认集群模式下slave是不对外提供服务的。核心源码如下

private void connect(List<String> endpoints, int index, Handler<AsyncResult<RedisConnection>> onConnect) {
    //如果RedisSlaves的值不是NEVER,就执行readonly
    connectionManager.getConnection(context, endpoints.get(index), RedisSlaves.NEVER != options.getUseSlave() ? cmd(READONLY) : null, getConnection -> {
    }
}

poolCleanerInterval

连接池空闲连接清理间隔，每次扫描时会直接将闲置的连接关闭。-1表示不开启空闲连接清理，核心源码在RedisConnectionManager#start()

synchronized void start() {
    long period = options.getPoolCleanerInterval();
    //延迟period时间后执行checkExpired()进行空闲连接清理
    this.timerID = period > 0 ? vertx.setTimer(period, id -> checkExpired(period)) : -1;
  }

poolRecycleTimeout

这个是用来控制连接池中连接在执行完命令后还能存活的时间，超过这个时间连接就会被关闭，核心源码在RedisStandaloneConnection。如下

public boolean isValid() {
    return expirationTimestamp > 0 && System.currentTimeMillis() <= expirationTimestamp;
  }

  @Override
  public void close() {
    // recycle this connection from the pool
    expirationTimestamp = recycleTimeout > 0 ? System.currentTimeMillis() + recycleTimeout : 0L;
    listener.onRecycle();
  }

maxPoolSize

连接池最大大小。只有当等待获取连接的请求数量达到maxPoolWaiting时才会创建新的连接，可以类比ThreadPoolExecutor的maximumPoolSize

maxPoolWaiting

连接池等待队列大小。当我们通过RedisClient或者RedisApi去操作reids时，如果当前已无空闲的redis连接，那么这个请求就会进入连接池的等待队列中。可以类比ThreadPoolExecutor的workQueue

maxWaitingHandlers

等待执行的handler数的最大值。当我们发送完redis指令后，一般是需要对这个响应作出处理，这个过程会被包装成一个task添加到eventLoop的waitQueue中，这项配置控制的就是这个queue的size

RedisClient

redisClient本质是对上述redisConnection的再封装，提供了针对redisServer不同模式的不同处理，分为3种：单机，哨兵以及集群模式。对应到具体的class：RedisClient、RedisSentinelClient、RedisSentinelClient
接下来来看一下如何配置这些客户端

STANDALONE

这个是redisClient默认的一种配置模式，现在我们用前面描述的redisOptions实践一下。

    RedisOptions options = new RedisOptions();
    options.setType(RedisClientType.STANDALONE)
            //指定server地址
            .setConnectionString("redis://ip:port")
            .setMaxPoolSize(6)
            .setMaxWaitingHandlers(1024)
            .setPoolRecycleTimeout(15_000)
            .setMaxPoolWaiting(2);

    Redis client = Redis.createClient(Vertx.vertx(), options);
    //创建redis 指令. get
    Request command = Request.cmd(Command.GET);
    //添加指令参数aa. 最后到达redis的指令就是get aa
    command.arg("aa");
    //发送指令并指定一个匿名handler处理结果
    client.send(command).onComplete(event -> {
        System.out.println("the value of redis key 'aa' is"+event.result().toString());
    });

SENTINEL

这里请原谅我秀了一下我那蹩脚的英语水平...

    RedisOptions options = new RedisOptions();
    options.setType(RedisClientType.SENTINEL)
            //the master name which are monitored by sentinels
            .setMasterName("myMaster")
            .setRole(RedisRole.MASTER)
            .setMaxPoolSize(6)
            .setMaxWaitingHandlers(1024)
            .setPoolRecycleTimeout(15_000)
            .setMaxPoolWaiting(2);

    //这里只填哨兵的地址就行了
    //程序会自动从哨兵处获取节点信息
    List<String> sentinels = new ArrayList<>();
    sentinels.add("redis://ip:port");
    sentinels.add("redis://ip:port");
    options.setEndpoints(sentinels);
    
    ...

这里可以只填一个哨兵地址，由于client不会主动去连接监控同一个master的哨兵节点，所以一旦这个唯一的哨兵挂了，那么server就会处于不可用的状态。

CLUSTER

    RedisOptions options = new RedisOptions();
    options.setType(RedisClientType.CLUSTER)
            .setUseSlave(RedisSlaves.NEVER)
            .setMaxPoolSize(6)
            .setMaxWaitingHandlers(1024)
            .setPoolRecycleTimeout(15_000)
            .setMaxPoolWaiting(20);
    //添加集群节点，越多越好...
    List<String> clusters = new ArrayList<>();
    sentinels.add("redis://ip:port");
    sentinels.add("redis://ip:port");
    sentinels.add("redis://ip:port");
    options.setEndpoints(clusters);

这里也需要注意最好将cluster中所有的节点信息给添加进来。虽然连接任意一个节点，client都会通过执行cluster slots获取cluster中所有节点的信息，但是长久保持连接的只有我们在这里设置进去的节点。这意味着一旦我们手动设置的节点挂了，那么server对我们来说就不可用了

RedisApi

这个是基于RedisClient的封装，隐藏了Command的复杂细节，提供了一套可以直接调用的api。使用如下：

    Redis client = Redis.createClient(Vertx.vertx(), options);
    RedisAPI api = RedisAPI.api(client);
    api.get("aa").onComplete(event -> {
        System.out.println("the value of redis key 'aa' is" + event.result().toString());
    });

总结

全文描述了vertx-client是什么以及其相对某些redis-client所具备的特点。同时深挖了这个客户端各项配置的作用以及不同模式下的表现。
总的来说，这篇文章可以帮助大家去建立一个基于vertx的redis客户端，并且对这个客户端能够做到基本的知其然且知其所以然。