Vert.x Redis

RedisOptions options = new RedisOptions();
options.setConnectionString("redis://:abracadabra@localhost:6379/1");
NetClientOptions tcpOptions = options.getNetClientOptions();
tcpOptions
  .setSsl(true)
  .setTrustOptions(new PemTrustOptions().addCertPath("/path/to/server.crt"))
  // 算法可以设为空字符串 ""，或者设为 "HTTPS" 以验证服务器主机名
  .setHostnameVerificationAlgorithm(algo);
Redis.createClient(
    vertx,
    options)
  .connect()
  .onSuccess(conn -> {
    // 使用该连接
  });

redis://[:password@]host[:port][/db-number]

unix://[:password@]/domain/docker.sock[?select=db-number]

public class RedisConnectOptionsSupplier<OPTS extends RedisConnectOptions> implements Supplier<Future<OPTS>> {
  private final Vertx vertx;
  private final RedisOptions options;
  private final Function<RedisOptions, OPTS> creator;
  private final Supplier<String> userName;
  private final Supplier<String> password;
  private final AtomicReference<Future<OPTS>> future;

  public RedisConnectOptionsSupplier(Vertx vertx, RedisOptions options, Function<RedisOptions, OPTS> creator,
    Supplier<String> userName, Supplier<String> password) {
    this.vertx = vertx;
    this.options = options;
    this.creator = creator;
    this.userName = userName;
    this.password = password;
    this.future = new AtomicReference<>();
  }

  @Override
  public Future<OPTS> get() {
    while (true) {
      Future<OPTS> currentFuture = this.future.get();
      if (currentFuture != null) {
        return currentFuture;
      }

      Promise<OPTS> promise = Promise.promise();
      Future<OPTS> future = promise.future();
      if (this.future.compareAndSet(null, future)) {
        OPTS result = creator.apply(options);
        result.setUser(userName.get());
        result.setPassword(password.get()); // TODO 在 Worker 线程中执行
        promise.complete(result);

        // 10 分钟后清理，以强制刷新
        vertx.setTimer(10 * 60 * 1000, ignored -> {
          this.future.set(null);
        });

        return future;
      }
    }
  }
}

package examples;

import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.http.SdkHttpMethod;
import software.amazon.awssdk.http.SdkHttpRequest;
import software.amazon.awssdk.http.auth.aws.signer.AwsV4FamilyHttpSigner;
import software.amazon.awssdk.http.auth.aws.signer.AwsV4HttpSigner;
import software.amazon.awssdk.http.auth.spi.signer.SignRequest;

import java.net.URI;
import java.time.Duration;

public class IamAuthToken {
  private static final String PROTOCOL = "http://";

  private final String userId;
  private final String replicationGroupId;
  private final String region;
  private final AwsCredentialsProvider credentials;

  public IamAuthToken(String userId, String replicationGroupId, String region, AwsCredentialsProvider credentials) {
    this.userId = userId;
    this.replicationGroupId = replicationGroupId;
    this.region = region;
    this.credentials = credentials;
  }

  public String getUserId() {
    return userId;
  }

  public String getToken() {
    URI uri = URI.create(PROTOCOL + replicationGroupId + "/");
    SdkHttpRequest request = SdkHttpRequest.builder()
      .method(SdkHttpMethod.GET)
      .uri(uri)
      .appendRawQueryParameter("Action", "connect")
      .appendRawQueryParameter("User", userId)
      .build();

    SdkHttpRequest signedRequest = sign(request, credentials.resolveCredentials());
    return signedRequest.getUri().toString().replace(PROTOCOL, "");
  }

  private SdkHttpRequest sign(SdkHttpRequest request, AwsCredentials credentials) {
    SignRequest<AwsCredentials> signRequest = SignRequest.builder(credentials)
      .request(request)
      .putProperty(AwsV4HttpSigner.REGION_NAME, region)
      .putProperty(AwsV4HttpSigner.SERVICE_SIGNING_NAME, "elasticache")
      .putProperty(AwsV4HttpSigner.EXPIRATION_DURATION, Duration.ofSeconds(900))
      .putProperty(AwsV4HttpSigner.AUTH_LOCATION, AwsV4FamilyHttpSigner.AuthLocation.QUERY_STRING)
      .build();
    return AwsV4HttpSigner.create().sign(signRequest).request();
  }
}

AwsCredentialsProvider credentialsProvider = DefaultCredentialsProvider.builder()
  .asyncCredentialUpdateEnabled(true)
  .build();
IamAuthToken token = new IamAuthToken("my-user", "my-redis", "us-east-1", credentialsProvider);

Redis client = Redis.createStandaloneClient(vertx, redisOptions, new RedisConnectOptionsSupplier<>(vertx,
  redisOptions, RedisStandaloneConnectOptions::new, token::getUserId, token::getToken));

Redis.createClient(
    vertx,
    new RedisOptions()
      .setType(RedisClientType.REPLICATION)
      .setTopology(RedisTopology.STATIC)
      .addConnectionString("redis://localhost:7000")
      .addConnectionString("redis://localhost:7004")
      .setMaxPoolSize(4)
      .setMaxPoolWaiting(16))
  .connect()
  .onSuccess(conn -> {
    // 这是一个复制模式客户端，
    // 写操作将路由至 Master 节点
    conn.send(Request.cmd(Command.SET).arg("key").arg("value"));
    // 读操作可能路由至 Replica 节点
    // （取决于具体配置及可用状态）
    conn.send(Request.cmd(Command.GET).arg("key"));
  });

options.setClusterTransactions(RedisClusterTransactions.SINGLE_NODE);

Redis redis = Redis.createClient(vertx);
redis.connect()
  .onSuccess(conn -> {
    conn.handler(EventBusHandler.create(vertx));
    conn.send(Request.cmd(Command.SUBSCRIBE).arg("news"));
  });

{
  "status": "OK",
  "type": "message|subscribe|unsubscribe|pmessage|psubscribe|punsubscribe",
  "value": {
    "channel": "<the channel>", (1)
    "message": "<the message>", (2)
    "pattern": "<the pattern>", (3)
    "current": <number of current subscriptions> (4)
  }
}