Bulk REPLY OFF Loading w/ Jedipus

JedipusBulkReplyOffLoading.java

private static int runWithJedipus(final IntStream keys) {

  int count = 0; 

  try (final RedisClient client =
      RedisClientFactory.startBuilding().create(Node.create("127.0.0.1", 6379))) {

    client.sendCmd(Cmds.FLUSHALL);
    
    final long start = System.currentTimeMillis();

    client.replyOff();

    final byte[] defaultValue = RESP.toBytes("D");

    count =
        (int) keys.peek(key -> client.sendCmd(Cmds.SETNX, RESP.toBytes(key), defaultValue)).count();

    client.replyOn();

    System.out.format("Loading %d key/values took %dms.%n", count,
        System.currentTimeMillis() - start); 

    final long dbsize = client.sendCmd(Cmds.DBSIZE.prim());
    if (dbsize != count) {
      throw new IllegalStateException(
          String.format("Expected %d keys, but %d exist.", count, dbsize));
    }
  }

  return count;
}

JedisBulkPipelinedLoading.java

private static int runWithJedis(final IntStream keys) {

  final int[] cnt = {0}; 

  try (final Jedis jedis = new Jedis("127.0.0.1", 6379)) {

    jedis.flushAll();

    final long start = System.currentTimeMillis();

    final byte[] defaultValue = RESP.toBytes("D");
    final Pipeline pipeline = jedis.pipelined();

    keys.forEach(key -> {
      if (++cnt[0] % 1_000 == 0) {
        pipeline.sync();
      }
      pipeline.setnx(RESP.toBytes(key), defaultValue);
    });
    pipeline.sync();

    System.out.format("Loading %d key/values took %dms.%n", cnt[0],
        System.currentTimeMillis() - start);

    final long dbsize = jedis.dbSize();
    if (dbsize != cnt[0]) {
      throw new IllegalStateException(
          String.format("Expected %d keys, but %d exist.", cnt[0], dbsize));
    }
  }

  return cnt[0];
}

Give this a go:

private int runWithJedipus(final BufferedReader br) {

  try (final RedisClient client =
      RedisClientFactory.startBuilding().create(Node.create(redisServerHost, redisServerPort))) {

    if (isFlushEnabled && !isFlushed) {
      client.sendCmd(Cmds.FLUSHALL);
      isFlushed = true;
    }

    final byte[] defaultValue = RESP.toBytes(DEFAULT_VALUE);    

    client.replyOff();

    return (int) br.lines()
        .map(RESP::toBytes)
        .peek(key -> client.sendCmd(Cmds.SETNX, key, defaultValue)).count();
  }
}

Please make sure you are on at least version '2.6.17', I only mention it, because I often use dynamic versions, e.g. '2.6.+', and they won't update if their cache duration has not expired.

My intention with changing the output buffer size was for further experimentation. I'm not sure what the optimal buffer size is for this application.

Np, since I'm using maven it's ok to get that updated quickly. I'll try ( and later one create an test on jedipus with jmh so we can test that within the lib)

@dgomesbr, I've tuned the output buffer size in the latest release. For the examples in this gist I'm seeing a 15-20% improvement over Jedis. The improvement will be even greater when accessing a remote Redis server.