Databases
API docs
You are currently viewing the documentation for the unreleased version 5.0.0.CR2 of Vert.x. Visit the latest stable version of this page.

Vert.x Redis

Vert.x Redis is a Redis client to be used with Vert.x.

This module allows data to be saved, retrieved, searched for, and deleted in a Redis. Redis is an open source, advanced key-value store. It is often referred to as a data structure server since keys can contain strings, hashes, lists, sets and sorted sets. To use this module you must have a Redis server instance running on your network.

Redis has a rich API and it can be organized in the following groups:

  • Cluster - Commands related to cluster management, note that using most of these commands you will need a Redis server with version >=3.0.0

  • Connection - Commands that allow you to switch DBs, connect, disconnect and authenticate to a server.

  • Hashes - Commands that allow operations on hashes.

  • HyperLogLog - Commands to approximating the number of distinct elements in a multiset, a HyperLogLog.

  • Keys - Commands to work with Keys.

  • List - Commands to work with Lists.

  • Pub/Sub - Commands to create queues and pub/sub clients.

  • Scripting - Commands to run Lua Scripts in Redis.

  • Server - Commands to manage and get server configurations.

  • Sets - Commands to work with un ordered sets.

  • Sorted Sets - Commands to work with sorted sets.

  • Strings - Commands to work with Strings.

  • Transactions - Commands to handle transaction lifecycle.

  • Streams - Commands to handle streaming.

Using Vert.x Redis

To use the Vert.x Redis client, add the following dependency to the dependencies section of your build descriptor.

  • Maven (in your pom.xml):

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-redis-client</artifactId>
  <version>${vertx-redis.version}</version>
</dependency>

  • Gradle (in your build.gradle):

compile 'io.vertx:vertx-redis-client:${vertx-redis.version}'

Connecting to Redis

The Redis client can operate in 4 distinct modes:

  • Standalone client (probably what most users need).

  • Sentinel (when working with Redis in High Availability mode).

  • Cluster (when working with Redis in Clustered mode).

  • Replication (single shard, one node writes, multiple read).

The connection mode is selected by the factory method on the Redis interface. Regardless of the mode, the client can be configured using a RedisOptions data object. By default, some configuration values are initialized with the following values:

  • netClientOptions: default is TcpKeepAlive: true, TcpNoDelay: true

  • endpoint: default is redis://localhost:6379

  • masterName: default is mymaster

  • role: default is MASTER

  • useReplicas: default is NEVER

In order to obtain a connection use the following code:

Redis.createClient(vertx)
  .connect()
  .onSuccess(conn -> {
    // use the connection
  });

In the configuration contains a password and/or a select database, these 2 commands will be executed automatically once a successful connection is established to the server.

Redis.createClient(
    vertx,
    // The client handles REDIS URLs. The select database as per spec is the
    // numerical path of the URL and the password is the password field of
    // the URL authority
    "redis://:abracadabra@localhost:6379/1")
  .connect()
  .onSuccess(conn -> {
    // use the connection
  });

Connecting using TLS

You can connect to a Redis server using TLS by configuring the client TCP options, make sure to set:

  • the ssl flag

  • the server certificate or the trust all flag

  • the hostname verification algorithm to "HTTPS" if you want to verify the server identity; otherwise ""

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"))
  // Algo can be the empty string "" or "HTTPS" to verify the server hostname
  .setHostnameVerificationAlgorithm(algo);
Redis.createClient(
    vertx,
    options)
  .connect()
  .onSuccess(conn -> {
    // use the connection
  });
More details on the TLS client config can be found here.

Connection String

The client will recognize addresses that follow the expression:

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

Or

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

When specifying a password or a database, those commands are always executed on connection start.

Providing RedisConnectOptions asynchronously

The Redis.createClient() method takes a single RedisOptions object that contains all options. This is the most common way of connecting to Redis.

However, there’s also an option to provide RedisOptions synchronously and the RedisConnectOptions asynchronously. There are 4 methods with the same parameter list that allow this:

  • Redis.createStandaloneClient()

  • Redis.createReplicationClient()

  • Redis.createSentinelClient()

  • Redis.createClusterClient()

These methods accept the Vertx object, the RedisOptions object, and a Supplier<Future<RedisConnectOptions>>. The RedisOptions object mainly provides NetClientOptions and PoolOptions, which are static. The Supplier<Future<RedisConnectOptions>> is used whenever a connection needs to be created and provides dynamic options. The type clearly shows that these dynamic options may be provided asynchronously.

The prime example when you might want this is when using Amazon ElastiCache with IAM authentication. The IAM authentication accepts short-lived tokens (their lifetime is only 15 minutes), so they need to be regenerated frequently.

Here’s an implementation of the Supplier<Future<RedisConnectOptions>> that caches the RedisConnectOptions for 10 minutes:

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 run on worker thread
        promise.complete(result);

        // clean up in 10 minutes, to force refresh
        vertx.setTimer(10 * 60 * 1000, ignored -> {
          this.future.set(null);
        });

        return future;
      }
    }
  }
}

To create the token, here’s a helper class that’s heavily based on https://github.com/aws-samples/elasticache-iam-auth-demo-app/blob/main/src/main/java/com/amazon/elasticache/IAMAuthTokenRequest.java:

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();
  }
}

This helper class might be instantiated like this:

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

Then, the Redis client might be instantiated like this:

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

Running commands

Given that the Redis client is connected to the server, all commands are now possible to execute using this module. The module offers a clean API for executing commands without the need to handwrite the command itself, for example if one wants to get a value of a key it can be done as:

RedisAPI redis = RedisAPI.api(client);

redis
  .get("mykey")
  .onSuccess(value -> {
    // do something...
  });

The response object is a generic type that allows converting from the basic Redis types to your language types. For example, if your response is of type INTEGER then you can get the value as any numeric primitive type int, long, etc.

Or you can perform more complex tasks such as handling responses as iterators:

if (response.type() == ResponseType.MULTI) {
  for (Response item : response) {
    // do something with item...
  }
}

Sentinel mode

To work with the sentinel mode (also known as high availability), the connection creation is quite similar:

Redis.createClient(
    vertx,
    new RedisOptions()
      .setType(RedisClientType.SENTINEL)
      .addConnectionString("redis://127.0.0.1:5000")
      .addConnectionString("redis://127.0.0.1:5001")
      .addConnectionString("redis://127.0.0.1:5002")
      .setMasterName("sentinel7000")
      .setRole(RedisRole.MASTER))
  .connect()
  .onSuccess(conn -> {
    conn.send(Request.cmd(Command.INFO))
      .onSuccess(info -> {
        // do something...
      });
  });

The connection strings here point to the sentinel nodes, which are used to discover the actual master and replica nodes.

What is important to notice is that in this mode, when the selected role is MASTER (which is the default) and when automatic failover is enabled (RedisOptions.setAutoFailover(true)), there is an extra connection to one of the sentinels that listens for failover events. When the sentinel notifies that a new master was elected, all clients will close their connection to the old master and transparently reconnect to the new master.

Note that there is a brief period of time between the old master failing and the new master being elected when the existing connections will temporarily fail all operations. After the new master is elected, the connections will automatically switch to it and start working again.

Cluster mode

To work with cluster, the connection creation is quite similar:

final RedisOptions options = new RedisOptions()
  .setType(RedisClientType.CLUSTER)
  .addConnectionString("redis://127.0.0.1:7000")
  .addConnectionString("redis://127.0.0.1:7001")
  .addConnectionString("redis://127.0.0.1:7002")
  .addConnectionString("redis://127.0.0.1:7003")
  .addConnectionString("redis://127.0.0.1:7004")
  .addConnectionString("redis://127.0.0.1:7005");

In this case, the configuration requires one or more members of the cluster to be known. This list will be used to ask the cluster for the current configuration, which means if any of the listed members is not available, it will be skipped.

In cluster mode, a connection is established to each node and special care is needed when executing commands. It is recommended to read the Redis manual in order to understand how clustering works. The client operating in this mode will do a best effort to identify which slot is used by the executed command in order to execute it on the right node. There could be cases where this isn’t possible to identify and in that case, as a best effort, the command will be run on a random node.

To know which Redis node holds which slots, the clustered Redis client holds a cache of the hash slot assignment. When the cache is empty, the first attempt to acquire a connection will execute CLUSTER SLOTS. The cache has a configurable TTL (time to live), which defaults to 1 second. The cache is also cleared whenever any command executed by the client receives the MOVED redirection.

Replication Mode

Working with replication is transparent to the client. Acquiring a connection is an expensive operation. The client will loop the provided endpoints until the master node is found. Once the master node is identified (this is the node where all write commands will be executed) a best effort is done to connect to all replica nodes (the read nodes).

With all node knowledge the client will now filter operations that perform read or writes to the right node type. Note that the useReplica configuration affects this choice. Just like with clustering, when the configuration states that the use of replica nodes is ALWAYS then any read operation will be performed on a replica node, SHARED will randomly share the read between master and replicas and finally NEVER means that replicas are never to be used.

The recommended usage of this mode, given the connection acquisition cost, is to re-use the connection as long as the application may need it.

Redis.createClient(
    vertx,
    new RedisOptions()
      .setType(RedisClientType.REPLICATION)
      .addConnectionString("redis://localhost:7000")
      .setMaxPoolSize(4)
      .setMaxPoolWaiting(16))
  .connect()
  .onSuccess(conn -> {
    // this is a replication client,
    // write operations will end up on the master node
    conn.send(Request.cmd(Command.SET).arg("key").arg("value"));
    // and read operations may end up on the replica nodes
    // (depending on configuration and their availability)
    conn.send(Request.cmd(Command.GET).arg("key"));
  });

Static topology

The replication mode allows configuring the multi-node topology statically. With static topology, the first node in the configuration is assumed to be a master node, while the remaining nodes are assumed to be replicas. The nodes are not verified; it is a responsibility of the application developer to ensure that the static configuration is correct.

To do this:

  • call RedisOptions.addConnectionString() repeatedly to configure the static topology (the first call configures the master node, subsequent calls configure replica nodes), and

  • call RedisOptions.setTopology(RedisTopology.STATIC).

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 -> {
    // this is a replication client,
    // write operations will end up on the master node
    conn.send(Request.cmd(Command.SET).arg("key").arg("value"));
    // and read operations may end up on the replica nodes
    // (depending on configuration and their availability)
    conn.send(Request.cmd(Command.GET).arg("key"));
  });

Note that automatic discovery of the topology is usually the preferred choice. Static configuration should only be used when necessary. One such case is Amazon Elasticache for Redis (Cluster Mode Disabled), where:

  • master node should be set to the primary endpoint, and

  • one replica node should be set to the reader endpoint.

Note that the reader endpoint of Elasticache for Redis (Cluster Mode Disabled) is a domain name which resolves to a CNAME record that points to one of the replicas. The CNAME record to which the reader endpoint resolves changes over time. This form of DNS-based load balancing does not work well with DNS resolution caching and connection pooling. As a result, some replicas are likely to be underutilized. Elasticache for Redis (Cluster Mode Enabled) doesn’t suffer from this problem, because it uses classic round-robin DNS.

Redis transactions

The Vert.x Redis client supports Redis transactions. You simply have to issue the corresponding commands: MULTI, EXEC, DISCARD, WATCH or UNWATCH. Note that transactions in Redis are not classic ACID transactions from SQL databases; they merely allow queueing multiple commands for later execution.

Transactions must be executed on a single connection. Trying to execute a transactional command in a connection-less mode (Redis.send()) will fail. It is possible to execute a transaction in a connection-less batch (Redis.batch()), but the batch must contain the entire transaction; it must not be split in multiple batches.

It is recommended to always obtain a connection (Redis.connect()) and execute all commands of a transaction on that connection.

Transactions in cluster

By default, transactions in Redis cluster are disabled. Attempting to execute a transactional command leads to a failure.

It is possible to enable single-node transactions in Redis cluster by:

options.setClusterTransactions(RedisClusterTransactions.SINGLE_NODE);

In single-node transactions, the first command (if it is WATCH) or the second command (if the first one is MULTI) determines on which node the transaction should execute. The connection is bound to the selected node and all subsequent commands are sent to that node, regardless of the hash slot assignment. When the final command of a transaction (EXEC or DISCARD) is executed, the connection is reset to default mode and is no longer bound to a single node.

If the transaction starts with WATCH, that command has keys and so determines the target node. If the transaction starts with MULTI, that command is not sent to Redis directly but is rather queued until the next command is executed. It is that command that determines the target node (so it should have keys, otherwise the target node is random).

Note that all this only applies to RedisConnection.send(). Command batches (RedisConnection.batch()) are always executed on a single node in the cluster, so there is no special support for transactions (they are not even disabled by default). Again, the batch must contain the entire transaction; it must not be split in multiple batches.

Pub/Sub mode

Redis supports queues and pub/sub mode, when operated in this mode once a connection invokes a subscriber mode then it cannot be used for running other commands than the command to leave that mode.

To start a subscriber one would do:

Redis.createClient(vertx, new RedisOptions())
  .connect()
  .onSuccess(conn -> {
    conn.handler(message -> {
      // do whatever you need to do with your message
    });
  });

And from another place in the code publish messages to the queue:

redis.send(Request.cmd(Command.PUBLISH).arg("channel1").arg("Hello World!"))
  .onSuccess(res -> {
    // published!
  });
It is important to remember that the commands SUBSCRIBE, UNSUBSCRIBE, PSUBSCRIBE and PUNSUBSCRIBE are void. This means that the result in case of success is null not a instance of response. All messages are then routed through the handler on the client.

EventBusHandler

The Vert.x Redis client version 4.x automatically forwarded messages to the Vert.x event bus unless a RedisConnection.handler() was registered.

In Vert.x Redis client version 5.x, that automatic forwarding is gone. If you still want it, you have to manually create an instance of EventBusHandler and register it using RedisConnection.handler():

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

The EventBusHandler allows customizing the address prefix, so if you want to use an address of com.example.<the channel> (instead of io.vertx.redis.<the channel>), you can use EventBusHandler.create(vertx, "com.example").

The message sent to the Vert.x event bus is a JsonObject with the following format:

{
  "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)
  }
}
1 For [p]message, subscribe and unsubscribe.
2 For [p]message.
3 For pmessage, psubscribe and punsubscribe.
4 For [p]subscribe and [p]unsubscribe.

The event bus address is <prefix>.<the channel> for message, subscribe and unsubscribe messages, and <prefix>.<the pattern> for pmessage, psubscribe and punsubscribe messages.

Tracing commands

The Redis client can trace command execution when Vert.x has tracing enabled.

The client reports a client span with the following details:

  • operation name: Command

  • tags:

    • db.user: the database username, if set

    • db.instance: the database number, if known (typically 0)

    • db.statement: the Redis command, without arguments (e.g. get or set)

    • db.type: redis

The default tracing policy is PROPAGATE, the client will only create a span when involved in an active trace.

You can change the client policy with setTracingPolicy, e.g you can set ALWAYS to always report a span:

options.setTracingPolicy(TracingPolicy.ALWAYS);

Domain Sockets

Most of the examples shown connecting to a TCP sockets, however it is also possible to use Redis connecting to a UNIX domain docket:

Redis.createClient(vertx, "unix:///tmp/redis.sock")
  .connect()
  .onSuccess(conn -> {
    // so something...
  });

Be aware that HA and cluster modes report server addresses always on TCP addresses not domain sockets. So the combination is not possible. Not because of this client but how Redis works.

Connection Pooling

All client variations are backed by a connection pool. By default, the configuration sets the pool size to 1, which means that it operates just like a single connection. There are 4 tunables for the pool:

  • maxPoolSize the max number of connections on the pool (default 6)

  • maxPoolWaiting the max waiting handlers to get a connection on a queue (default 24)

  • poolCleanerInterval the interval how often connections will be cleaned (default 30 seconds)

  • poolRecycleTimeout the timeout to keep an unused connection in the pool (default 3 mintues)

Pooling is quite useful to avoid custom connection management, for example you can just use as:

Redis.createClient(vertx, "redis://localhost:7006")
  .send(Request.cmd(Command.PING))
  .onSuccess(res -> {
    // Should have received a pong...
  });

It is important to observe that no connection was acquired or returned, it’s all handled by the pool. However, there might be some scalability issues when more than 1 concurrent request attempts to get a connection from the pool; in order to overcome this, we need to tune the pool. A common configuration is to set the maximum size of the pool to the number of available CPU cores and allow requests to get a connection from the pool to queue:

Redis.createClient(
    vertx,
    new RedisOptions()
      .setConnectionString("redis://localhost:7006")
      // allow at max 8 connections to redis
      .setMaxPoolSize(8)
      // allow 32 connection requests to queue waiting
      // for a connection to be available.
      .setMaxWaitingHandlers(32))
  .send(Request.cmd(Command.PING))
  .onSuccess(res -> {
    // Should have received a pong...
  });
Pooling is not compatible with SUBSCRIBE, UNSUBSCRIBE, PSUBSCRIBE or PUNSUBSCRIBE, because these commands will modify the way the connection operates and the connection cannot be reused.

Implementing Reconnect on Error

While the connection pool is quite useful, for performance, a connection should not be auto managed but controlled by you. In this case you will need to handle connection recovery, error handling and reconnect.

A typical scenario is that a user will want to reconnect to the server whenever an error occurs. The automatic reconnect is not part of the redis client as it will force a behaviour that might not match the user expectations, for example:

  1. What should happen to current in-flight requests?

  2. Should the exception handler be invoked or not?

  3. What if the retry will also fail?

  4. Should the previous state (db, authentication, subscriptions) be restored?

  5. Etc…​

In order to give the user full flexibility, this decision should not be performed by the client. However, a simple reconnect with backoff timeout could be implemented as follows:

class RedisVerticle extends VerticleBase {

  private static final int MAX_RECONNECT_RETRIES = 16;

  private final RedisOptions options = new RedisOptions();
  private Redis redis;
  private RedisConnection client;
  private final AtomicBoolean CONNECTING = new AtomicBoolean();

  @Override
  public Future<?> start() {
    return createRedisClient()
      .onSuccess(conn -> {
        // connected to redis!
      });
  }

  /**
   * Will create a redis client and setup a reconnect handler when there is
   * an exception in the connection.
   */
  private Future<RedisConnection> createRedisClient() {
    Promise<RedisConnection> promise = Promise.promise();

    // make sure to invalidate old connection if present
    if (redis != null) {
      redis.close();;
    }

    if (CONNECTING.compareAndSet(false, true)) {
      redis = Redis.createClient(vertx, options);
      redis
        .connect()
        .onSuccess(conn -> {
          client = conn;

          // make sure the client is reconnected on error
          // eg, the underlying TCP connection is closed but the client side doesn't know it yet
          //     the client tries to use the staled connection to talk to server. An exceptions will be raised
          conn.exceptionHandler(e -> {
            attemptReconnect(0);
          });

          // make sure the client is reconnected on connection close
          // eg, the underlying TCP connection is closed with normal 4-Way-Handshake
          //     this handler will be notified instantly
          conn.endHandler(placeHolder -> {
            attemptReconnect(0);
          });

          // allow further processing
          promise.complete(conn);
          CONNECTING.set(false);
        }).onFailure(t -> {
          promise.fail(t);
          CONNECTING.set(false);
        });
    } else {
      promise.complete();
    }

    return promise.future();
  }

  /**
   * Attempt to reconnect up to MAX_RECONNECT_RETRIES
   */
  private void attemptReconnect(int retry) {
    if (retry > MAX_RECONNECT_RETRIES) {
      // we should stop now, as there's nothing we can do.
      CONNECTING.set(false);
    } else {
      // retry with backoff up to 10240 ms
      long backoff = (long) (Math.pow(2, Math.min(retry, 10)) * 10);

      vertx.setTimer(backoff, timer -> {
        createRedisClient()
          .onFailure(t -> attemptReconnect(retry + 1));
      });
    }
  }
}

In this example, the client object will be replaced on reconnect and the application will retry up to 16 times with a backoff up to 1280ms. By discarding the client we ensure that all old inflight responses are lost and all new ones will be on the new connection.

It is important to note that the reconnect will create a new connection object, so these object references should not be cached and evaluated every time.

Protocol Parser

This client supports both RESP2 and RESP3 protocols. By default, the client attempts to negotiate support for RESP3 at connection handshake time.

It is possible to use the setPreferredProtocolVersion method to select the preferred version, RESP2 or RESP3:

options.setPreferredProtocolVersion(ProtocolVersion.RESP2);

The parser internally creates an "infinite" readable buffer from all the chunks received from the server, in order to avoid creating too much garbage in terms of memory collection, a tunable watermark value is configurable at JVM startup time. The system property io.vertx.redis.parser.watermark defines how much data is kept in this readable buffer before it gets discarded. By default, this value is 16 KB. This means that each connection to the server will use at least this amount of memory. As the client works in pipeline mode, keeping the number of connections low provides best results, which means 16 KB * nconn memory will be used. If the application will require a large number of connections, then reducing the watermark value to a smaller value or even disable it entirely is advisable.