Redis Pipeline: spring-data-redis vs simple jedis

application.yml

server:
    port: 8080

spring:
    profiles:
        active: dev
    redis:
        host: 127.0.0.1
        port: 6379

RedisLoaderApplication.java

import com.lambdaworks.redis.RedisAsyncConnection;
import com.lambdaworks.redis.RedisClient;
import com.lambdaworks.redis.SetArgs;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Pipeline;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

@SpringBootApplication
public class RedisLoaderApplication implements CommandLineRunner {
    /**
     * Default value for keys inserted
     */
    public static String DEFAULT_VALUE = "1";
    /**
     * http://redis.io/commands/SET - NX -- Only set the key if it does not already exist.
     */
    public static String NX_MODE = "NX";
    /**
     * Used for syncing operations with Jedis
     */
    public AtomicLong counter = new AtomicLong(0L);
    Logger logger = org.slf4j.LoggerFactory.getLogger(RedisLoaderApplication.class);
    @Value("${loader.redis.ip:127.0.0.1}")
    String redisServerHost;
    @Value("${loader.redis.port:6379}")
    int redisServerPort;
    @Value("${loader.input.src}")
    String inputFile;
    @Value("${loader.input.method:JEDIS}")
    String runWith;
    /**
     * To be used along with inputFile
     */
    @Value("${loader.input.pattern:*}")
    String inputFilePattern;

    public static void main(String[] args) {
        SpringApplication.run(RedisLoaderApplication.class, args);
    }

    JedisPool jedisPool() {
        int defaultTimeout = 30;

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setTestOnBorrow(true);
        poolConfig.setTestWhileIdle(true);
        poolConfig.setMaxWaitMillis(defaultTimeout);
        poolConfig.setMaxTotal(8);
        poolConfig.setMinIdle(8);
        return new JedisPool(poolConfig, redisServerHost, redisServerPort, defaultTimeout);
    }

    private void runWithJedis(BufferedReader br) {
        JedisPool pool = jedisPool();
        try (Jedis jedis = pool.getResource()) {
            jedis.flushAll();
            Pipeline p = jedis.pipelined();
            br.lines().forEach(key -> {
                p.set(key, DEFAULT_VALUE, NX_MODE);

                if (counter.addAndGet(1L) == 350_000) {
                    counter.set(0L);
                    p.sync();
                }
            });
            p.sync();
        }
    }

    private void runWithLettuce(BufferedReader br) {
        RedisClient client = RedisClient.create("redis://" + redisServerHost + ":" + redisServerPort + "/0");

        /*
        *  The AutoFlushCommands state is set per connection and affects therefore all threads using
        *  the shared connection. If you want to omit this effect, use dedicated connections.
        *  The [!!!]AutoFlushCommands state cannot be set on pooled connections by the lettuce connection pooling.
        * */
        //RedisConnectionPool<RedisAsyncCommands<String, String>> pool = client.asyncPool();

        //try(RedisAsyncCommands<String, String> cn = pool.allocateConnection()){
        RedisAsyncConnection<String, String> cn = client.connectAsync();
        cn.flushall();
        cn.setAutoFlushCommands(false);
        br.lines().forEach(key -> {
            cn.set(key, DEFAULT_VALUE, SetArgs.Builder.nx());

            if (counter.addAndGet(1L) == 350_000) {
                counter.set(0L);
                cn.flushCommands();
            }
        });
        cn.flushCommands();
        //}

        //pool.close();
    }

    @Override
    public void run(String... args) throws Exception {
        logger.info("Redis Loader started");
        long start = System.nanoTime();

        List<Path> files = FileUtils.listFiles(Paths.get("/load/"), "(^x.*)");


        // test data
        BufferedReader br = new BufferedReader(
                new InputStreamReader(
                        new FileInputStream(inputFile), StandardCharsets.UTF_8
                ),
                64 * 1000 * 1000
        );

        if ("JEDIS".equalsIgnoreCase(runWith)) {
            // 10_000_000 in 49s [475_000 control sync]
            // 10_000_000 in 29s [350_000 control sync] <<< USED, less sync calls
            // 10_000_000 in 28s [275_000 control sync]
            // 10_000_000 in 28s [200_000 control sync]
            // 10_000_000 in 27s [100_000 control sync]
            // 10_000_000 in 27s [ 50_000 control sync]
            runWithJedis(br);
        } else {
            // ---------------------------------------------------------------------
            // *DEAD* restTemplate doesn't support chunking, would require chunking
            // before the file loop
            // ---------------------------------------------------------------------
            // 10_000_000 in 512s [Using Reader outside the file]
            // x in -s [OOM (probably cause of the SYNC for 10kk responses)]
            // runWithRedisTemplate(br);


            /*
            * Unexpected exception during request: java.lang.OutOfMemoryError: Java heap space java.lang.OutOfMemoryError: Java heap space
            * */
            runWithLettuce(br);
        }

        logger.info("Redis Loader finished (took " + TimeUnit.SECONDS.convert(System.nanoTime() - start, TimeUnit.NANOSECONDS) + "s)");
    }
}

sample-data.input

$ head LOAD.txt

5511946258960_41_0
555384433222_39_0
555384535698_82_0
5589999183267_74_0
5581987694232_41_0
5521989909934_82_0
554184720023_26_1431367
5581987365904_93_0
5511946769372_39_1543498
5511988257589_141_0

@xetorthio opened redis/jedis#1272 so we can see if that comes in handy to other people.

Please Let me know the resulta of tour debug.

@dgomesbr interesting case. Please keep us posted about the results so we can improve Jedis to handle these cases.
/cc @xetorthio @HeartSaVioR

@xetorthio @marcosnils
Seems like gist doesn't send notification mail when I'm mentioned.

@dgomesbr
I didn't know Java Socket supports full duplex. If reading and writing at the same time doesn't make an issue (Sorry I'm not expert on this), we could have sync thread to sync periodically in the background. I guess it could be used with normal case, not only 'fire and forget' case.

@xetorthio @marcosnils @marcosnils I've put all the info and the updated code at redis/jedis#1272 so we don't miss notification and interaction.

Debug shown nothing valuable, but I found that Redis on Windows is not as reliable for testing purposes as Linux. We couldn't reproduce it on Ubuntu