How to Benchmark Redis Stream Java APIs with Dynamic Threads and QPS

Background

Redis Stream commands (e.g., XADD, XREAD, XDEL) behave similarly to list operations. This article shows how to benchmark the add‑and‑delete pattern using Groovy scripts that interact with Redis via the Jedis client.

Test Scenarios

Three use cases were defined:

Add messages to a stream.

Read messages from a stream.

Add a message and immediately delete it.

The first two are straightforward and omitted from the detailed code. The third scenario is implemented with two dynamic load models: a dynamic‑thread model and a dynamic‑QPS (queries‑per‑second) model.

Dynamic Thread Model

All test parameters are declared as static fields for simplicity: RedisBase driver – wrapper around a Jedis connection (default 127.0.0.1:6379). String key = "FunTesterStream" – the stream name. Map<String,Integer> map – field/value pairs to store (e.g., {"key1":1, "key2":2}). XAddParams params – controls XADD options; initially set with .maxLen(1000) to cap stream length.

The core workload resides in com.funtest.groovytest.RedisStreamTest.FunTester.doing(). Each invocation adds a record and immediately deletes it using the returned redis.clients.jedis.StreamEntryID:

package com.funtest.groovytest;

import com.funtester.base.constaint.FunThread;
import com.funtester.db.redis.RedisBase;
import com.funtester.frame.SourceCode;
import com.funtester.frame.execute.FunConcurrent;
import redis.clients.jedis.params.XAddParams;
import java.util.concurrent.atomic.AtomicInteger;

public class RedisStreamTest extends SourceCode {
    static RedisBase driver;
    static AtomicInteger index = new AtomicInteger(0);
    static String desc = "Redis stream performance test practice";
    static Map<String,Integer> map;
    static String key = "FunTesterStream";
    static XAddParams params;

    public static void main(String[] args) {
        driver = new RedisBase("127.0.0.1", 6379);
        params = XAddParams.xAddParams().maxLen(1000);
        map = new HashMap<>();
        map.put("key1", 1);
        map.put("key2", 2);
        // Warm‑up entry
        driver.xadd(key, params, map);
        // Use default params for the test loop
        params = XAddParams.xAddParams();
        new FunConcurrent(new FunTester()).start();
    }

    private static class FunTester extends FunThread {
        FunTester() {
            super(null, desc + index.getAndIncrement());
        }
        @Override
        protected void doing() throws Exception {
            // Add a record and immediately delete it
            String entryId = driver.xadd(key, params, map);
            driver.xdel(key, entryId);
        }
        @Override
        public FunTester clone() {
            return new FunTester();
        }
    }
}

FunConcurrent

creates a configurable pool of threads (default can be overridden via its constructor) that repeatedly invoke doing(), allowing measurement of throughput and latency as thread count varies.

Dynamic QPS Model

The same add‑delete logic is wrapped in a closure that is scheduled by FunQpsConcurrent. This component accepts a target QPS value and invokes the closure at that rate, providing a more precise control over request intensity.

package com.funtest.groovytest;

import com.funtester.db.redis.RedisBase;
import com.funtester.frame.SourceCode;
import com.funtester.frame.execute.FunQpsConcurrent;
import redis.clients.jedis.params.XAddParams;
import java.util.HashMap;
import java.util.Map;

public class RedisStreamQPSTest extends SourceCode {
    static RedisBase driver;
    static String desc = "Redis stream performance test practice";
    static Map<String,Integer> map;
    static String key = "FunTesterStream";
    static XAddParams params;

    public static void main(String[] args) {
        driver = new RedisBase("127.0.0.1", 6379);
        params = XAddParams.xAddParams().maxLen(1000);
        map = new HashMap<>();
        map.put("key1", 1);
        map.put("key2", 2);
        // Warm‑up entry
        driver.xadd(key, params, map);
        // Use default params for the test loop
        params = XAddParams.xAddParams();
        // Closure that performs one add‑delete cycle
        Runnable test = () -> {
            String entryId = driver.xadd(key, params, map);
            driver.xdel(key, entryId);
        };
        // Start QPS‑controlled execution (default QPS can be set inside the class)
        new FunQpsConcurrent(test).start();
    }
}

Running the Tests and Sample Output

Compile the Groovy/Java sources and run the main method while a Redis server is listening on 127.0.0.1:6379. The console prints initialization details, daemon thread status, and periodic statistics such as:

Designed QPS vs. actual QPS.

Number of active threads.

Per‑thread efficiency (actual QPS ÷ designed QPS).

15:50:41.091 main
##### #      #  #    # ####### ######  #####  ####### ######  #####
#      #     #  ##   #    #    #       #         #    #       #    #
###   #      #  # #  #    #    ####    #####      #    ####    ####
#      #     #  #  # #    #    #            #    #    #       #   #
#       #####   #    #    ######  #####      #    ######  #    #
15:50:41.474 main redis connection pool IP:127.0.0.1, port:6379, timeout:5000
15:50:42.728 Deamon daemon thread started!
15:50:45.761 main designed QPS:1, actual QPS:0 active threads:0 single‑thread efficiency:0
15:50:50.780 main designed QPS:1, actual QPS:1 active threads:1 single‑thread efficiency:1
15:50:55.798 main designed QPS:1, actual QPS:1 active threads:0 single‑thread efficiency:1
Process ended, exit code 130 (interrupted by signal 2: SIGINT)

The output confirms that the test harness can achieve the configured request rate and reports thread activity in real time.

Conclusion

The two Groovy test programs provide a concrete template for benchmarking Redis Stream add‑delete operations under either a variable thread pool or a precise QPS controller. Developers can adapt the static parameters, field map, or stream key to match their own workloads and use the reported metrics (throughput, latency, thread efficiency) to evaluate Redis performance in their environments.