From 388512056a43eee7d9bd3b65bb54c498f7e101b7 Mon Sep 17 00:00:00 2001 From: Andy Stark Date: Fri, 30 Oct 2026 12:02:49 +0000 Subject: [PATCH 1/2] DOC-7226 Move four more Jedis use cases from JedisPool to RedisClient Prefetch cache, time-series dashboard, pub/sub and streaming (java-jedis) take a RedisClient instead of a JedisPool. As in the first batch, getResource() blocks are gone, pipelines and MULTI transactions are in try-with-resources, demo servers build the client with RedisClient.create or RedisClient.builder() plus a ConnectionPoolConfig where they size the pool, and each page's Maven/Gradle version, prose and javac/java commands move to Jedis 8.0.2 with the six runtime jars. The three directories that used APIs RedisClient lacks needed more than a rename: - time-series-dashboard reached for jedis.getConnection() and wrote raw commands to it. It now goes through jedis.executeCommand(new CommandArguments(cmd).addObjects(args)). UnifiedJedis.sendCommand(ProtocolCommand, String...) is the obvious swap, but Jedis 8 deprecates it, which would trade one deprecated call for another. - pub-sub used Jedis.pubsubChannels/pubsubNumSub/pubsubNumPat, which exist only on the legacy Jedis class. They are now executeCommand(new CommandObject<>( new CommandArguments(Protocol.Command.PUBSUB).add(Protocol.Keyword.X), BuilderFactory.Y)). Each Subscription no longer owns a Jedis; its worker thread calls jedis.subscribe(listener, ...) on the shared client, which borrows and holds a pooled connection until unsubscribe. The "close the Jedis to force subscribe() to return" fallback is removed because closing the shared client would be wrong. - streaming's XAUTOCLAIM raw-reply path moves from sendCommand to executeCommand, and StreamEntryID.UNRECEIVED_ENTRY (deprecated) becomes XREADGROUP_UNDELIVERED_ENTRY. The page's claim that the typed xautoclaim drops the deleted-ids slot still holds in 8.0.2 (checked with javap), so the hand parsing stays. Verification, compiled with -Xlint:deprecation (0 warnings) against Jedis 8.0.2 and run on redis:8.10. Prefetch: bulk load, sync-worker transaction, add, delete, 200 concurrent mixed calls. Time series: snapshot parsing matches the original on Jedis 5.2.0 (types and min<=avg<=max), 200 concurrent snapshots. Pub/sub: exact and pattern delivery counts exact (203 = 3 + 200), PUBSUB introspection correct, 150 subscribe/unsubscribe cycles against a pool of 64 (a leak would hang by ~60). Streaming: the same scenario on the original (5.2.0) and migrated builds gives identical output for produce, consume, crash, pending=3, XAUTOCLAIM claimed=3, add/remove worker, trim and replay; only the final concurrent-load tail differs, which is timing. Every page's javac/java commands were extracted and run literally in a clean directory with only the six jars; all four compile and serve. Constraint: RedisClient has no pubsubChannels/NumSub/NumPat, getConnection(), or non-deprecated sendCommand; use executeCommand(CommandArguments) or executeCommand(CommandObject) Constraint: a subscribe()/psubscribe() on a RedisClient holds one pooled connection for the subscription's life, so maxTotal must cover subscribers plus command traffic Learned: JedisPubSub needs redis.clients.authentication.core.Token (redis-authx-core) at class-load time, so a three-jar classpath serves every other demo and fails only pub/sub Learned: an unrelated JVM left deadlocked from an earlier probe held 8 pooled connections and inflated every later "clients: N" count; check pgrep before reading connection counts Learned: the streaming "crash" button drops the next N deliveries, so a scenario must crash before producing or nothing is ever pending Ticket: DOC-7226 Co-Authored-By: Claude Sonnet 5.5 --- .../prefetch-cache/java-jedis/DemoServer.java | 19 +- .../java-jedis/PrefetchCache.java | 97 ++++------ .../prefetch-cache/java-jedis/_index.md | 33 ++-- .../pub-sub/java-jedis/DemoServer.java | 19 +- .../pub-sub/java-jedis/RedisPubSubHub.java | 89 ++++----- .../use-cases/pub-sub/java-jedis/_index.md | 54 +++--- .../streaming/java-jedis/DemoServer.java | 26 +-- .../streaming/java-jedis/EventStream.java | 180 ++++++++---------- .../use-cases/streaming/java-jedis/_index.md | 112 ++++++----- .../java-jedis/DemoServer.java | 12 +- .../java-jedis/RedisTimeSeriesStore.java | 72 +++---- .../java-jedis/_index.md | 8 +- 12 files changed, 325 insertions(+), 396 deletions(-) diff --git a/content/develop/use-cases/prefetch-cache/java-jedis/DemoServer.java b/content/develop/use-cases/prefetch-cache/java-jedis/DemoServer.java index 186f7cf455..e06ba288c6 100644 --- a/content/develop/use-cases/prefetch-cache/java-jedis/DemoServer.java +++ b/content/develop/use-cases/prefetch-cache/java-jedis/DemoServer.java @@ -1,8 +1,7 @@ import com.sun.net.httpserver.HttpExchange; import com.sun.net.httpserver.HttpHandler; import com.sun.net.httpserver.HttpServer; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.RedisClient; import java.io.IOException; import java.io.InputStream; @@ -28,9 +27,9 @@ * without ever falling back to the primary on the read path.

* * 
{@code
- * javac -cp jedis-5.1.2.jar:slf4j-api-2.0.13.jar \
+ * javac -cp jedis-8.0.2.jar:commons-pool2-2.13.1.jar:slf4j-api-1.7.36.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar \
  *     PrefetchCache.java MockPrimaryStore.java SyncWorker.java DemoServer.java
- * java -cp .:jedis-5.1.2.jar:slf4j-api-2.0.13.jar \
+ * java -cp .:jedis-8.0.2.jar:commons-pool2-2.13.1.jar:slf4j-api-1.7.36.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar \
  *     DemoServer --port 8785 --redis-host localhost --redis-port 6379
  * }
*/ @@ -39,7 +38,7 @@ public class DemoServer { private static PrefetchCache cache; private static MockPrimaryStore primary; private static SyncWorker sync; - private static JedisPool jedisPool; + private static RedisClient jedis; public static void main(String[] args) { String host = "127.0.0.1"; @@ -79,14 +78,14 @@ public static void main(String[] args) { } try { - jedisPool = new JedisPool(new JedisPoolConfig(), redisHost, redisPort); - jedisPool.getResource().close(); + jedis = RedisClient.create(redisHost, redisPort); + jedis.ping(); } catch (Exception e) { System.err.printf("Failed to connect to Redis at %s:%d: %s%n", redisHost, redisPort, e.getMessage()); System.exit(1); } - cache = new PrefetchCache(jedisPool, cachePrefix, ttlSeconds); + cache = new PrefetchCache(jedis, cachePrefix, ttlSeconds); primary = new MockPrimaryStore(primaryLatencyMs); sync = new SyncWorker(primary, cache); @@ -121,12 +120,12 @@ public static void main(String[] args) { Runtime.getRuntime().addShutdownHook(new Thread(() -> { sync.stop(); server.stop(0); - jedisPool.close(); + jedis.close(); })); } catch (IOException e) { System.err.println("Failed to start server: " + e.getMessage()); sync.stop(); - jedisPool.close(); + jedis.close(); System.exit(1); } } diff --git a/content/develop/use-cases/prefetch-cache/java-jedis/PrefetchCache.java b/content/develop/use-cases/prefetch-cache/java-jedis/PrefetchCache.java index e323d5c5ee..c207685036 100644 --- a/content/develop/use-cases/prefetch-cache/java-jedis/PrefetchCache.java +++ b/content/develop/use-cases/prefetch-cache/java-jedis/PrefetchCache.java @@ -6,10 +6,9 @@ import java.util.Map; import java.util.concurrent.atomic.AtomicLong; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisPool; +import redis.clients.jedis.AbstractTransaction; import redis.clients.jedis.Pipeline; -import redis.clients.jedis.Transaction; +import redis.clients.jedis.RedisClient; import redis.clients.jedis.params.ScanParams; import redis.clients.jedis.resps.ScanResult; @@ -33,7 +32,7 @@ public class PrefetchCache { public static final String DEFAULT_PREFIX = "cache:category:"; public static final int DEFAULT_TTL_SECONDS = 3600; - private final JedisPool pool; + private final RedisClient jedis; private final String prefix; private final int ttlSeconds; @@ -45,18 +44,18 @@ public class PrefetchCache { private double syncLagMsTotal; private long syncLagSamples; - public PrefetchCache(JedisPool pool) { - this(pool, DEFAULT_PREFIX, DEFAULT_TTL_SECONDS); + public PrefetchCache(RedisClient jedis) { + this(jedis, DEFAULT_PREFIX, DEFAULT_TTL_SECONDS); } - public PrefetchCache(JedisPool pool, String prefix, int ttlSeconds) { - if (pool == null) { - throw new IllegalArgumentException("pool is required"); + public PrefetchCache(RedisClient jedis, String prefix, int ttlSeconds) { + if (jedis == null) { + throw new IllegalArgumentException("jedis is required"); } if (ttlSeconds < 1) { throw new IllegalArgumentException("ttlSeconds must be at least 1 second"); } - this.pool = pool; + this.jedis = jedis; this.prefix = (prefix == null || prefix.isEmpty()) ? DEFAULT_PREFIX : prefix; this.ttlSeconds = ttlSeconds; } @@ -96,8 +95,7 @@ public Result(Map record, boolean hit, double redisLatencyMs) { */ public int bulkLoad(Iterable> records) { int loaded = 0; - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); + try (Pipeline pipe = jedis.pipelined()) { for (Map record : records) { if (record == null) { continue; @@ -114,7 +112,7 @@ public int bulkLoad(Iterable> records) { } if (loaded > 0) { pipe.sync(); - } + } } synchronized (statsLock) { prefetched += loaded; @@ -133,9 +131,7 @@ public Result get(String entityId) { String cacheKey = cacheKey(entityId); long startedNs = System.nanoTime(); Map cached; - try (Jedis jedis = pool.getResource()) { - cached = jedis.hgetAll(cacheKey); - } + cached = jedis.hgetAll(cacheKey); double redisLatencyMs = (System.nanoTime() - startedNs) / 1_000_000.0; if (cached != null && !cached.isEmpty()) { @@ -185,17 +181,14 @@ public void applyChange(Map change) { // just drops it. return; } - try (Jedis jedis = pool.getResource()) { - Transaction tx = jedis.multi(); + try (AbstractTransaction tx = jedis.multi()) { tx.del(cacheKey); tx.hset(cacheKey, fields); tx.expire(cacheKey, ttlSeconds); tx.exec(); } } else if ("delete".equals(op)) { - try (Jedis jedis = pool.getResource()) { - jedis.del(cacheKey); - } + jedis.del(cacheKey); } else { return; } @@ -214,9 +207,7 @@ public void applyChange(Map change) { /** Delete one cache key. Demo-only: simulates a broken sync pipeline. */ public boolean invalidate(String entityId) { - try (Jedis jedis = pool.getResource()) { - return jedis.del(cacheKey(entityId)) == 1L; - } + return jedis.del(cacheKey(entityId)) == 1L; } /** Delete every key under this cache's prefix and return the count. */ @@ -224,18 +215,16 @@ public int clear() { int deleted = 0; String match = prefix + "*"; ScanParams params = new ScanParams().match(match).count(500); - try (Jedis jedis = pool.getResource()) { - String cursor = ScanParams.SCAN_POINTER_START; - do { - ScanResult scan = jedis.scan(cursor, params); - cursor = scan.getCursor(); - List keys = scan.getResult(); - if (keys != null && !keys.isEmpty()) { - // DEL accepts multiple keys; one round trip per batch. - deleted += (int) jedis.del(keys.toArray(new String[0])); - } - } while (!ScanParams.SCAN_POINTER_START.equals(cursor)); - } + String cursor = ScanParams.SCAN_POINTER_START; + do { + ScanResult scan = jedis.scan(cursor, params); + cursor = scan.getCursor(); + List keys = scan.getResult(); + if (keys != null && !keys.isEmpty()) { + // DEL accepts multiple keys; one round trip per batch. + deleted += (int) jedis.del(keys.toArray(new String[0])); + } + } while (!ScanParams.SCAN_POINTER_START.equals(cursor)); return deleted; } @@ -243,16 +232,14 @@ public int clear() { public List ids() { List result = new ArrayList<>(); ScanParams params = new ScanParams().match(prefix + "*").count(500); - try (Jedis jedis = pool.getResource()) { - String cursor = ScanParams.SCAN_POINTER_START; - do { - ScanResult scan = jedis.scan(cursor, params); - cursor = scan.getCursor(); - for (String key : scan.getResult()) { - result.add(stripPrefix(key)); - } - } while (!ScanParams.SCAN_POINTER_START.equals(cursor)); - } + String cursor = ScanParams.SCAN_POINTER_START; + do { + ScanResult scan = jedis.scan(cursor, params); + cursor = scan.getCursor(); + for (String key : scan.getResult()) { + result.add(stripPrefix(key)); + } + } while (!ScanParams.SCAN_POINTER_START.equals(cursor)); Collections.sort(result); return result; } @@ -260,21 +247,17 @@ public List ids() { public int count() { int n = 0; ScanParams params = new ScanParams().match(prefix + "*").count(500); - try (Jedis jedis = pool.getResource()) { - String cursor = ScanParams.SCAN_POINTER_START; - do { - ScanResult scan = jedis.scan(cursor, params); - cursor = scan.getCursor(); - n += scan.getResult().size(); - } while (!ScanParams.SCAN_POINTER_START.equals(cursor)); - } + String cursor = ScanParams.SCAN_POINTER_START; + do { + ScanResult scan = jedis.scan(cursor, params); + cursor = scan.getCursor(); + n += scan.getResult().size(); + } while (!ScanParams.SCAN_POINTER_START.equals(cursor)); return n; } public long ttlRemaining(String entityId) { - try (Jedis jedis = pool.getResource()) { - return jedis.ttl(cacheKey(entityId)); - } + return jedis.ttl(cacheKey(entityId)); } /** Hit/miss + sync counters with derived hit-rate and average sync lag. */ diff --git a/content/develop/use-cases/prefetch-cache/java-jedis/_index.md b/content/develop/use-cases/prefetch-cache/java-jedis/_index.md index d901e91649..5925f22e37 100644 --- a/content/develop/use-cases/prefetch-cache/java-jedis/_index.md +++ b/content/develop/use-cases/prefetch-cache/java-jedis/_index.md @@ -47,12 +47,11 @@ The `PrefetchCache` class wraps the cache operations ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/prefetch-cache/java-jedis/PrefetchCache.java)): ```java -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.RedisClient; -JedisPool pool = new JedisPool(new JedisPoolConfig(), "localhost", 6379); +RedisClient jedis = RedisClient.create("localhost", 6379); MockPrimaryStore primary = new MockPrimaryStore(80); -PrefetchCache cache = new PrefetchCache(pool); +PrefetchCache cache = new PrefetchCache(jedis); // Pre-load every primary record into Redis in one pipelined round trip. cache.bulkLoad(primary.listRecords()); @@ -95,8 +94,7 @@ The `bulkLoad` method pipelines a `DEL` + `HSET` + `EXPIRE` triple for every rec ```java public int bulkLoad(Iterable> records) { int loaded = 0; - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); + try (Pipeline pipe = jedis.pipelined()) { for (Map record : records) { if (record == null) continue; String entityId = record.get("id"); @@ -113,7 +111,7 @@ public int bulkLoad(Iterable> records) { } ``` -The pipeline is intentionally non-transactional on the **startup** path: nothing is reading the cache yet, the records do not need to be applied atomically as a set, and skipping `MULTI`/`EXEC` keeps the bulk load fast. The same method is used for the live `/reprefetch` reload, which is safe because the demo pauses the sync worker around the clear-and-reload sequence — see [Re-prefetch under load](#re-prefetch-under-load) below. If you call `bulkLoad` directly from your own code on a cache that is already serving reads, either pause your writers first or wrap the writes in a `Transaction` so callers cannot observe a half-loaded record. +The pipeline is intentionally non-transactional on the **startup** path: nothing is reading the cache yet, the records do not need to be applied atomically as a set, and skipping `MULTI`/`EXEC` keeps the bulk load fast. The same method is used for the live `/reprefetch` reload, which is safe because the demo pauses the sync worker around the clear-and-reload sequence — see [Re-prefetch under load](#re-prefetch-under-load) below. If you call `bulkLoad` directly from your own code on a cache that is already serving reads, either pause your writers first or wrap the writes in a transaction so callers cannot observe a half-loaded record. ## Reads from Redis only @@ -124,9 +122,7 @@ public Result get(String entityId) { String cacheKey = cacheKey(entityId); long startedNs = System.nanoTime(); Map cached; - try (Jedis jedis = pool.getResource()) { - cached = jedis.hgetAll(cacheKey); - } + cached = jedis.hgetAll(cacheKey); double redisLatencyMs = (System.nanoTime() - startedNs) / 1_000_000.0; if (cached != null && !cached.isEmpty()) { @@ -158,17 +154,14 @@ public void applyChange(Map change) { // Malformed upsert: skip rather than crash the sync worker. return; } - try (Jedis jedis = pool.getResource()) { - Transaction tx = jedis.multi(); + try (AbstractTransaction tx = jedis.multi()) { tx.del(cacheKey); tx.hset(cacheKey, fields); tx.expire(cacheKey, ttlSeconds); tx.exec(); } } else if ("delete".equals(op)) { - try (Jedis jedis = pool.getResource()) { - jedis.del(cacheKey); - } + jedis.del(cacheKey); } } ``` @@ -278,13 +271,13 @@ curl -O $BASE/DemoServer.java ### Start the demo server -From that directory, with the Jedis and `slf4j-api` jars in the working directory (the example assumes Jedis 5.1.2 and slf4j-api 2.0.13 — adjust the filenames to match your versions): +From that directory, with the Jedis jar and its dependencies in the working directory (the example assumes Jedis 8.0.2, commons-pool2 2.13.1, slf4j-api 1.7.36, gson 2.14.0, json 20260719, and redis-authx-core 0.1.1-beta2 — adjust the filenames to match your versions): ```bash -javac -cp jedis-5.1.2.jar:slf4j-api-2.0.13.jar \ +javac -cp jedis-8.0.2.jar:commons-pool2-2.13.1.jar:slf4j-api-1.7.36.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar \ PrefetchCache.java MockPrimaryStore.java SyncWorker.java DemoServer.java -java -cp .:jedis-5.1.2.jar:slf4j-api-2.0.13.jar \ +java -cp .:jedis-8.0.2.jar:commons-pool2-2.13.1.jar:slf4j-api-1.7.36.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar \ DemoServer --port 8785 --redis-host localhost --redis-port 6379 ``` @@ -372,9 +365,9 @@ Prefetch only works if the entire dataset fits in Redis memory with headroom. Es CDC pipelines are eventually consistent: an event can be lost (broker outage, consumer crash, configuration drift) and the cache can silently diverge from the source. Run a periodic reconciliation job that re-reads all primary records, compares them against the cache, and either re-prefetches or fixes individual entries. Even running it once a day catches drift that ad-hoc inspection would miss. -### Tune the JedisPool +### Tune the connection pool -The demo uses a default `JedisPoolConfig`. In production, set `setMaxTotal()` and `setMaxIdle()` to match your concurrency profile and server-side `maxclients`. Every cache call acquires a connection with try-with-resources so connections are returned to the pool even on exceptions; this is what lets the sync worker, the HTTP handlers, and any background job share Redis safely without an in-process lock. +The demo uses the default pool settings of `RedisClient.create()`. In production, build the client with `RedisClient.builder()` and a `ConnectionPoolConfig`, and set `setMaxTotal()` and `setMaxIdle()` to match your concurrency profile and server-side `maxclients`. Each cache command borrows a connection for the duration of the call and returns it, and the pipeline and transaction calls return theirs when try-with-resources closes them; this is what lets the sync worker, the HTTP handlers, and any background job share Redis safely without an in-process lock. An open pipeline or transaction holds a pooled connection, so send any reads it needs through the pipeline or transaction itself, not through the client. ### Namespace cache keys in shared Redis deployments diff --git a/content/develop/use-cases/pub-sub/java-jedis/DemoServer.java b/content/develop/use-cases/pub-sub/java-jedis/DemoServer.java index 7f589ab289..b5b72b85e8 100644 --- a/content/develop/use-cases/pub-sub/java-jedis/DemoServer.java +++ b/content/develop/use-cases/pub-sub/java-jedis/DemoServer.java @@ -1,8 +1,8 @@ import com.sun.net.httpserver.HttpExchange; import com.sun.net.httpserver.HttpHandler; import com.sun.net.httpserver.HttpServer; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.ConnectionPoolConfig; +import redis.clients.jedis.RedisClient; import java.io.IOException; import java.io.InputStream; @@ -36,7 +36,7 @@ public class DemoServer { ); private static RedisPubSubHub hub; - private static JedisPool jedisPool; + private static RedisClient jedis; public static void main(String[] args) { String host = "127.0.0.1"; @@ -63,19 +63,22 @@ public static void main(String[] args) { } } - JedisPoolConfig poolConfig = new JedisPoolConfig(); + ConnectionPoolConfig poolConfig = new ConnectionPoolConfig(); poolConfig.setMaxTotal(64); poolConfig.setMaxIdle(16); poolConfig.setMinIdle(2); - jedisPool = new JedisPool(poolConfig, redisHost, redisPort); - try (var jedis = jedisPool.getResource()) { + jedis = RedisClient.builder() + .hostAndPort(redisHost, redisPort) + .poolConfig(poolConfig) + .build(); + try { jedis.ping(); } catch (Exception e) { System.err.printf("Failed to connect to Redis at %s:%d: %s%n", redisHost, redisPort, e.getMessage()); System.exit(1); } - hub = new RedisPubSubHub(jedisPool); + hub = new RedisPubSubHub(jedis); seedDefaultSubscriptions(hub); HttpServer server; @@ -100,7 +103,7 @@ public static void main(String[] args) { } catch (Exception ignored) { } try { - jedisPool.close(); + jedis.close(); } catch (Exception ignored) { } server.stop(0); diff --git a/content/develop/use-cases/pub-sub/java-jedis/RedisPubSubHub.java b/content/develop/use-cases/pub-sub/java-jedis/RedisPubSubHub.java index 45306abe57..5002b09ea5 100644 --- a/content/develop/use-cases/pub-sub/java-jedis/RedisPubSubHub.java +++ b/content/develop/use-cases/pub-sub/java-jedis/RedisPubSubHub.java @@ -1,6 +1,9 @@ -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisPool; +import redis.clients.jedis.BuilderFactory; +import redis.clients.jedis.CommandArguments; +import redis.clients.jedis.CommandObject; import redis.clients.jedis.JedisPubSub; +import redis.clients.jedis.Protocol; +import redis.clients.jedis.RedisClient; import java.util.ArrayDeque; import java.util.ArrayList; @@ -70,7 +73,7 @@ public Map toMap() { * A named in-process subscriber bound to one or more channels or patterns. * *

Each {@code Subscription} owns its own {@link JedisPubSub} listener - * running on a dedicated thread with a dedicated {@link Jedis} + * running on a dedicated thread with a dedicated pooled * connection. Subscriptions are independent: closing one does not * affect another, even if they share channels.

*/ @@ -85,7 +88,6 @@ public final class Subscription { private long received = 0; private volatile boolean closed = false; - private final Jedis jedis; private final JedisPubSub listener; private final Thread thread; @@ -98,10 +100,10 @@ public final class Subscription { this.isPattern = isPattern; this.bufferSize = bufferSize; - // Each subscriber gets its own Jedis connection. Sharing one - // connection across subscribers would couple their lifetimes — - // unsubscribing one would close the channel for the others. - this.jedis = pool.getResource(); + // Each subscribe() call borrows its own pooled connection for as + // long as the subscription lasts. Sharing one connection across + // subscribers would couple their lifetimes — unsubscribing one + // would close the channel for the others. // ackLatch is counted down by Jedis once per target after Redis // acknowledges the SUBSCRIBE / PSUBSCRIBE. The constructor blocks @@ -141,18 +143,12 @@ public void onPSubscribe(String pattern, int subscribedChannels) { jedis.subscribe(listener, targetArr); } } catch (Exception e) { - // Connection closed by close() is expected on shutdown. // If we never got the first ack through, surface the // failure so the constructor's wait can give up. startupError.compareAndSet(0L, 1L); while (ackLatch.getCount() > 0) { ackLatch.countDown(); } - } finally { - try { - jedis.close(); - } catch (Exception ignored) { - } } }, "pubsub-" + name); this.thread.setDaemon(true); @@ -234,9 +230,8 @@ public void resetReceived() { *

Jedis ships {@code UNSUBSCRIBE}/{@code PUNSUBSCRIBE} back over * the same socket; after the listener has unsubscribed from every * target the blocking {@code subscribe}/{@code psubscribe} call - * returns and the thread exits. As a belt-and-braces guard we also - * close the {@link Jedis} connection in the {@code finally} block - * of the worker thread, which unblocks any read still in flight.

+ * returns, the thread exits, and the connection goes back to the + * pool.

*/ public void close() { if (closed) { @@ -258,19 +253,6 @@ public void close() { } catch (InterruptedException e) { Thread.currentThread().interrupt(); } - if (thread.isAlive()) { - // Last resort: dropping the connection forces the - // blocking subscribe() call to return. - try { - jedis.close(); - } catch (Exception ignored) { - } - try { - thread.join(500); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } } public boolean isAlive() { @@ -288,7 +270,7 @@ public Map toMap() { } } - private final JedisPool pool; + private final RedisClient jedis; private final int bufferSize; private final Map subscriptions = new ConcurrentHashMap<>(); @@ -298,12 +280,12 @@ public Map toMap() { private final AtomicLong deliveredTotal = new AtomicLong(); private final Map channelPublished = new ConcurrentHashMap<>(); - public RedisPubSubHub(JedisPool pool) { - this(pool, 50); + public RedisPubSubHub(RedisClient jedis) { + this(jedis, 50); } - public RedisPubSubHub(JedisPool pool, int bufferSize) { - this.pool = pool; + public RedisPubSubHub(RedisClient jedis, int bufferSize) { + this.jedis = jedis; this.bufferSize = bufferSize; } @@ -318,10 +300,7 @@ public RedisPubSubHub(JedisPool pool, int bufferSize) { */ public int publish(String channel, Object message) { String payload = JsonUtil.toJson(message); - long delivered; - try (Jedis jedis = pool.getResource()) { - delivered = jedis.publish(channel, payload); - } + long delivered = jedis.publish(channel, payload); publishedTotal.incrementAndGet(); deliveredTotal.addAndGet(delivered); channelPublished.merge(channel, 1L, Long::sum); @@ -374,11 +353,13 @@ public Subscription getSubscription(String name) { /** List server-side channels with at least one subscriber (PUBSUB CHANNELS). */ public List activeChannels(String pattern) { - try (Jedis jedis = pool.getResource()) { - List channels = new ArrayList<>(jedis.pubsubChannels(pattern == null ? "*" : pattern)); - Collections.sort(channels); - return channels; - } + List channels = new ArrayList<>(jedis.executeCommand(new CommandObject<>( + new CommandArguments(Protocol.Command.PUBSUB) + .add(Protocol.Keyword.CHANNELS) + .add(pattern == null ? "*" : pattern), + BuilderFactory.STRING_LIST))); + Collections.sort(channels); + return channels; } /** @@ -392,21 +373,23 @@ public Map channelSubscriberCounts(List channels) { if (channels == null || channels.isEmpty()) { return out; } - try (Jedis jedis = pool.getResource()) { - Map result = jedis.pubsubNumSub(channels.toArray(new String[0])); - // Preserve the input ordering so the JSON output is stable. - for (String channel : channels) { - out.put(channel, result.getOrDefault(channel, 0L)); - } + Map result = jedis.executeCommand(new CommandObject<>( + new CommandArguments(Protocol.Command.PUBSUB) + .add(Protocol.Keyword.NUMSUB) + .addObjects((Object[]) channels.toArray(new String[0])), + BuilderFactory.PUBSUB_NUMSUB_MAP)); + // Preserve the input ordering so the JSON output is stable. + for (String channel : channels) { + out.put(channel, result.getOrDefault(channel, 0L)); } return out; } /** Total active pattern subscriptions across all clients (PUBSUB NUMPAT). */ public long patternSubscriberCount() { - try (Jedis jedis = pool.getResource()) { - return jedis.pubsubNumPat(); - } + return jedis.executeCommand(new CommandObject<>( + new CommandArguments(Protocol.Command.PUBSUB).add(Protocol.Keyword.NUMPAT), + BuilderFactory.LONG)); } /** Combined publish and subscribe counters plus the current registry size. */ diff --git a/content/develop/use-cases/pub-sub/java-jedis/_index.md b/content/develop/use-cases/pub-sub/java-jedis/_index.md index cb1c1284d6..bcad397c45 100644 --- a/content/develop/use-cases/pub-sub/java-jedis/_index.md +++ b/content/develop/use-cases/pub-sub/java-jedis/_index.md @@ -29,14 +29,14 @@ That gives you: * Live server-side introspection through `PUBSUB CHANNELS`, `PUBSUB NUMSUB`, and `PUBSUB NUMPAT` * At-most-once delivery: subscribers that are offline when a message is published miss it, so durable state should live in keys or a Stream, not in the pub/sub channel itself -In this example, the publisher side calls `PUBLISH` with a JSON-encoded body and counts how many subscribers Redis reported delivering to. Each in-process subscriber owns its own `Jedis` connection from the pool and a dedicated worker thread that pumps incoming messages into a buffer. +In this example, the publisher side calls `PUBLISH` with a JSON-encoded body and counts how many subscribers Redis reported delivering to. Each in-process subscriber holds its own pooled connection and a dedicated worker thread that pumps incoming messages into a buffer. ## How it works The flow looks like this: 1. The application calls `hub.subscribe(name, channels)` or `hub.psubscribe(name, patterns)` -2. The helper acquires a dedicated `Jedis` connection from the pool, builds a `JedisPubSub` listener, and starts a daemon thread that calls `jedis.subscribe(listener, channels...)` (or `psubscribe(...)`) — both calls block for the lifetime of the subscription +2. The helper builds a `JedisPubSub` listener and starts a daemon thread that calls `jedis.subscribe(listener, channels...)` (or `psubscribe(...)`) on the shared `RedisClient` — each call borrows a dedicated connection from the pool and blocks for the lifetime of the subscription 3. The application (or another process) calls `hub.publish(channel, message)` 4. Redis fans the message out over every subscribing client's open socket 5. The listener's `onMessage` (or `onPMessage`) callback fires on the subscriber thread, wraps the raw message as a `ReceivedMessage`, and appends it to a per-subscriber ring buffer @@ -52,12 +52,12 @@ The `RedisPubSubHub` class wraps the publish, subscribe, and introspection opera ```java import java.util.List; import java.util.Map; -import redis.clients.jedis.JedisPool; +import redis.clients.jedis.RedisClient; public class Main { public static void main(String[] args) { - JedisPool pool = new JedisPool("localhost", 6379); - RedisPubSubHub hub = new RedisPubSubHub(pool); + RedisClient jedis = RedisClient.create("localhost", 6379); + RedisPubSubHub hub = new RedisPubSubHub(jedis); // Exact-match subscriber hub.subscribe("orders-listener", List.of("orders:new")); @@ -121,10 +121,7 @@ The implementation uses: ```java public int publish(String channel, Object message) { String payload = JsonUtil.toJson(message); - long delivered; - try (Jedis jedis = pool.getResource()) { - delivered = jedis.publish(channel, payload); - } + long delivered = jedis.publish(channel, payload); publishedTotal.incrementAndGet(); deliveredTotal.addAndGet(delivered); channelPublished.merge(channel, 1L, Long::sum); @@ -136,7 +133,7 @@ The integer returned by `PUBLISH` is what Redis itself reports — the number of ## Subscribing to channels -`subscribe()` creates a named in-process `Subscription` that owns its own `Jedis` connection and a dedicated worker thread: +`subscribe()` creates a named in-process `Subscription` that holds its own pooled connection and a dedicated worker thread: ```java public Subscription subscribe(String name, List channels) { @@ -147,8 +144,6 @@ public Subscription subscribe(String name, List channels) { Inside the `Subscription` constructor, the helper builds a `JedisPubSub` listener, starts the dispatch thread, and blocks until Redis has acknowledged every `SUBSCRIBE` so the very first publish can't race ahead: ```java -this.jedis = pool.getResource(); - final CountDownLatch ackLatch = new CountDownLatch(targets.size()); this.listener = new JedisPubSub() { @@ -181,7 +176,7 @@ if (!ackLatch.await(2, TimeUnit.SECONDS)) { A few details matter here: * `jedis.subscribe(...)` and `jedis.psubscribe(...)` **block the calling thread** until the listener unsubscribes from every target. Running each subscription on its own thread is the only way to keep the rest of the application responsive — there is no "run in background" helper in Jedis. -* Each `Subscription` gets its own `Jedis` connection from the pool. A connection that has issued `SUBSCRIBE` or `PSUBSCRIBE` is switched into subscribe-only mode, so other commands on the same connection would fail. Sharing one connection across subscribers would also couple their lifetimes — unsubscribing one would tear down the channel for the others. +* Each `subscribe()` or `psubscribe()` call borrows its own connection from the pool and keeps it until the subscription ends. A connection that has issued `SUBSCRIBE` or `PSUBSCRIBE` is switched into subscribe-only mode, so other commands on the same connection would fail — `RedisClient` runs those commands on other pooled connections. Sharing one connection across subscribers would also couple their lifetimes — unsubscribing one would tear down the channel for the others. * The constructor waits on a `CountDownLatch` that's decremented once per target inside `onSubscribe` / `onPSubscribe`. Without it, the parent thread could return from `subscribe()` while the spawned worker thread is still about to send the SUBSCRIBE command — and a `PUBLISH` issued immediately afterwards would beat the subscription into Redis. * The `JedisPubSub` listener is the same object on both threads: it's safe to call `listener.unsubscribe()` from a different thread because Jedis ships the unsubscribe command back over the listener's own socket. After the unsubscribe round-trip completes, the blocking `subscribe()` call returns and the worker thread exits. @@ -217,7 +212,7 @@ hub.patternSubscriberCount(); // PUBSUB NUMPAT `PUBSUB CHANNELS` only reports channels with at least one exact-match subscriber — pattern subscribers do not appear here. That's a deliberate Redis design choice: a glob like `*` would otherwise show up as a subscriber to every conceivable channel. `PUBSUB NUMPAT` covers the pattern side as a single global count. -Each of these commands acquires a regular (non-subscribe) `Jedis` connection from the pool with try-with-resources, so the bookkeeping calls never compete with the dedicated subscriber connections. +Each of these commands borrows a regular (non-subscribe) connection from the pool for the duration of the call, so the bookkeeping calls never use the dedicated subscriber connections. `RedisClient` has no `pubsubChannels()`, `pubsubNumSub()`, or `pubsubNumPat()` methods, so the helper sends these three commands with `executeCommand`. ## Stats and history @@ -247,7 +242,7 @@ public Map stats() { * Redis 6.2 or later running locally on the default port (6379). Earlier versions still work for plain `PUBLISH`/`SUBSCRIBE`; `PUBSUB NUMPAT` is older than that. * Java 17 or later. -* Jedis 5.x on the classpath. The smallest workable classpath is the Jedis jar plus its two transitive dependencies, `commons-pool2` and `slf4j-api`. +* Jedis 8.x on the classpath, along with its runtime dependencies: `commons-pool2`, `slf4j-api`, `gson`, `json`, and `redis-authx-core`. If you use Maven: @@ -255,14 +250,14 @@ If you use Maven: redis.clients jedis - 5.1.2 + 8.0.2 ``` If you use Gradle: ```groovy -implementation 'redis.clients:jedis:5.1.2' +implementation 'redis.clients:jedis:8.0.2' ``` ## Running the demo @@ -279,23 +274,26 @@ curl -O $BASE/DemoServer.java curl -O $BASE/JsonUtil.java ``` -You also need the three jars on the classpath. Grab them from [Maven Central](https://search.maven.org/) — search for `jedis`, `commons-pool2`, and `slf4j-api`, then download each jar from the artefact page — or copy them out of an existing local `~/.m2/repository` checkout if you've used these libraries before: +You also need the six jars on the classpath. Grab them from [Maven Central](https://search.maven.org/) — search for `jedis`, `commons-pool2`, `slf4j-api`, `gson`, `json`, and `redis-authx-core`, then download each jar from the artefact page — or copy them out of an existing local `~/.m2/repository` checkout if you've used these libraries before: -* `jedis-5.1.2.jar` -* `commons-pool2-2.12.0.jar` -* `slf4j-api-2.0.13.jar` +* `jedis-8.0.2.jar` +* `commons-pool2-2.13.1.jar` +* `slf4j-api-1.7.36.jar` +* `gson-2.14.0.jar` +* `json-20260719.jar` +* `redis-authx-core-0.1.1-beta2.jar` -(Any 5.x Jedis release works; the transitive versions of `commons-pool2` and `slf4j-api` are pinned by the Jedis POM, so use the versions listed in the Jedis 5.1.2 [Maven page](https://search.maven.org/artifact/redis.clients/jedis/5.1.2/jar) if you want an exact match. Other recent patch versions of either library work fine.) +(The versions of the other five jars are pinned by the Jedis POM, so use the versions listed in the Jedis 8.0.2 [Maven page](https://search.maven.org/artifact/redis.clients/jedis/8.0.2/jar) if you want an exact match.) ### Start the demo server A local demo server is included to show the hub in action ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/pub-sub/java-jedis/DemoServer.java)). Compile and run with `javac` and `java`, listing each jar on the classpath: ```bash -javac -cp jedis-5.1.2.jar:commons-pool2-2.12.0.jar:slf4j-api-2.0.13.jar \ +javac -cp jedis-8.0.2.jar:commons-pool2-2.13.1.jar:slf4j-api-1.7.36.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar \ JsonUtil.java RedisPubSubHub.java DemoServer.java -java -cp .:jedis-5.1.2.jar:commons-pool2-2.12.0.jar:slf4j-api-2.0.13.jar \ +java -cp .:jedis-8.0.2.jar:commons-pool2-2.13.1.jar:slf4j-api-1.7.36.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar \ DemoServer --port 8098 ``` @@ -323,11 +321,11 @@ If your Redis server is running elsewhere, start the demo with `--redis-host` an A subscriber that's offline when a message is published misses it permanently. For events you can't afford to lose, write the durable record (the order row, the cache key version, the audit log entry) to its primary store, then `PUBLISH` a notification so live consumers can pick it up immediately. On reconnect, consumers reconcile by reading the durable store, not by waiting for missed pub/sub messages. If you actually need replay or at-least-once delivery, switch to [Redis Streams](/content/develop/data-types/streams/_index.md) with consumer groups. -### Give every subscriber its own Jedis connection +### Give every subscriber its own connection -A `Jedis` instance that has called `subscribe(...)` or `psubscribe(...)` is in subscribe-only mode: any other command on the same connection will fail. Always pull the subscriber's connection from the pool with `pool.getResource()` and dedicate it to that subscriber until it's torn down. The helper does this for you — every `Subscription` opens a fresh connection and owns it for the subscription's lifetime — but if you write your own subscriber code, the same rule applies. +A connection that has issued `SUBSCRIBE` or `PSUBSCRIBE` is in subscribe-only mode: any other command on the same connection will fail. Each `subscribe(...)` or `psubscribe(...)` call on a `RedisClient` borrows a connection from the pool and dedicates it to that subscription until it's torn down. The helper does this for you — every `Subscription` makes its own call on its own thread and holds its connection for the subscription's lifetime — but if you write your own subscriber code, the same rule applies. -A related corollary: never let the pool size cap your subscriber count. If you have 100 long-lived subscribers in one process, `JedisPoolConfig.setMaxTotal(...)` needs to allow at least 100 connections for the subscribers plus whatever your publishers and bookkeeping calls need. Set it generously; idle subscriber connections cost almost nothing on the Redis side. +A related corollary: never let the pool size cap your subscriber count. If you have 100 long-lived subscribers in one process, `ConnectionPoolConfig.setMaxTotal(...)` needs to allow at least 100 connections for the subscribers plus whatever your publishers and bookkeeping calls need. Set it generously; idle subscriber connections cost almost nothing on the Redis side. ### Unsubscribe through the listener, not the connection @@ -349,7 +347,7 @@ The demo caps each subscriber's in-memory message buffer at 50. That's right for ### Sharded pub/sub on a Redis Cluster -On a Redis Cluster, plain `PUBLISH` fans every message out to every node via the cluster bus, which becomes a hotspot at high throughput. Redis 7.0 added [sharded pub/sub](/content/develop/pubsub/_index.md#sharded-pubsub): channels are hashed to slots, and `SPUBLISH` / `SSUBSCRIBE` only touch the shard that owns the slot. Jedis 5.x exposes `jedis.spublish(...)` and a `JedisShardedPubSub` listener subclass. If you're scaling pub/sub on a cluster, prefer the sharded commands and pick channel names whose hash distribution matches your traffic. +On a Redis Cluster, plain `PUBLISH` fans every message out to every node via the cluster bus, which becomes a hotspot at high throughput. Redis 7.0 added [sharded pub/sub](/content/develop/pubsub/_index.md#sharded-pubsub): channels are hashed to slots, and `SPUBLISH` / `SSUBSCRIBE` only touch the shard that owns the slot. Jedis 8.x exposes `spublish(...)` and `ssubscribe(...)` on its cluster client (`RedisClusterClient`) and a `JedisShardedPubSub` listener subclass. If you're scaling pub/sub on a cluster, prefer the sharded commands and pick channel names whose hash distribution matches your traffic. ### Inspect pub/sub state directly in Redis diff --git a/content/develop/use-cases/streaming/java-jedis/DemoServer.java b/content/develop/use-cases/streaming/java-jedis/DemoServer.java index dd0f2e3992..bbd1992c39 100644 --- a/content/develop/use-cases/streaming/java-jedis/DemoServer.java +++ b/content/develop/use-cases/streaming/java-jedis/DemoServer.java @@ -20,8 +20,8 @@ import java.util.concurrent.Executors; import java.util.concurrent.locks.ReentrantLock; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.ConnectionPoolConfig; +import redis.clients.jedis.RedisClient; /** * Redis streaming demo server (Jedis + JDK HttpServer). @@ -43,7 +43,7 @@ public class DemoServer { DEFAULT_GROUPS.put("analytics", Arrays.asList("worker-c")); } - private static JedisPool jedisPool; + private static RedisClient jedis; private static EventStream stream; private static StreamingDemo demo; private static final Random random = new Random(); @@ -90,20 +90,24 @@ public static void main(String[] args) { } try { - JedisPoolConfig config = new JedisPoolConfig(); + ConnectionPoolConfig config = new ConnectionPoolConfig(); // Sized for: 3 workers (blocking XREADGROUP) + 16 HTTP handlers - // + small headroom. Each blocking call holds its Jedis for the - // block duration, so undersizing the pool would starve handlers. + // + small headroom. Each blocking call holds a pooled connection + // for the block duration, so undersizing the pool would starve + // handlers. config.setMaxTotal(64); config.setMaxIdle(32); - jedisPool = new JedisPool(config, redisHost, redisPort); - jedisPool.getResource().close(); + jedis = RedisClient.builder() + .hostAndPort(redisHost, redisPort) + .poolConfig(config) + .build(); + jedis.ping(); } catch (Exception e) { System.err.printf("Failed to connect to Redis at %s:%d: %s%n", redisHost, redisPort, e.getMessage()); System.exit(1); } - stream = new EventStream(jedisPool, streamKey, maxlen, claimIdleMs); + stream = new EventStream(jedis, streamKey, maxlen, claimIdleMs); demo = new StreamingDemo(stream); if (resetOnStart) { @@ -140,12 +144,12 @@ public static void main(String[] args) { Runtime.getRuntime().addShutdownHook(new Thread(() -> { demo.stopAll(); server.stop(0); - jedisPool.close(); + jedis.close(); })); } catch (IOException e) { System.err.println("Failed to start server: " + e.getMessage()); demo.stopAll(); - jedisPool.close(); + jedis.close(); System.exit(1); } } diff --git a/content/develop/use-cases/streaming/java-jedis/EventStream.java b/content/develop/use-cases/streaming/java-jedis/EventStream.java index c0d397a828..f8eae856dc 100644 --- a/content/develop/use-cases/streaming/java-jedis/EventStream.java +++ b/content/develop/use-cases/streaming/java-jedis/EventStream.java @@ -4,9 +4,9 @@ import java.util.List; import java.util.Map; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisPool; +import redis.clients.jedis.CommandArguments; import redis.clients.jedis.Pipeline; +import redis.clients.jedis.RedisClient; import redis.clients.jedis.Response; import redis.clients.jedis.StreamEntryID; import redis.clients.jedis.commands.ProtocolCommand; @@ -57,7 +57,7 @@ public class EventStream { public static final int DEFAULT_MAXLEN_APPROX = 10_000; public static final int DEFAULT_CLAIM_MIN_IDLE_MS = 15_000; - private final JedisPool pool; + private final RedisClient jedis; private final String streamKey; private final int maxlenApprox; private final int claimMinIdleMs; @@ -67,15 +67,15 @@ public class EventStream { private long ackedTotal; private long claimedTotal; - public EventStream(JedisPool pool) { - this(pool, DEFAULT_STREAM_KEY, DEFAULT_MAXLEN_APPROX, DEFAULT_CLAIM_MIN_IDLE_MS); + public EventStream(RedisClient jedis) { + this(jedis, DEFAULT_STREAM_KEY, DEFAULT_MAXLEN_APPROX, DEFAULT_CLAIM_MIN_IDLE_MS); } - public EventStream(JedisPool pool, String streamKey, int maxlenApprox, int claimMinIdleMs) { - if (pool == null) { - throw new IllegalArgumentException("pool is required"); + public EventStream(RedisClient jedis, String streamKey, int maxlenApprox, int claimMinIdleMs) { + if (jedis == null) { + throw new IllegalArgumentException("jedis is required"); } - this.pool = pool; + this.jedis = jedis; this.streamKey = (streamKey == null || streamKey.isEmpty()) ? DEFAULT_STREAM_KEY : streamKey; this.maxlenApprox = maxlenApprox > 0 ? maxlenApprox : DEFAULT_MAXLEN_APPROX; @@ -137,9 +137,8 @@ public String produce(String eventType, Map payload) { */ public List produceBatch(List>> events) { List ids = new ArrayList<>(events.size()); - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); - List> responses = new ArrayList<>(events.size()); + List> responses = new ArrayList<>(events.size()); + try (Pipeline pipe = jedis.pipelined()) { for (Map.Entry> event : events) { Map fields = encodeFields(event.getKey(), event.getValue()); XAddParams params = new XAddParams() @@ -148,10 +147,10 @@ public List produceBatch(List>> ev responses.add(pipe.xadd(streamKey, params, fields)); } pipe.sync(); - for (Response resp : responses) { - StreamEntryID id = resp.get(); - ids.add(id == null ? null : id.toString()); - } + } + for (Response resp : responses) { + StreamEntryID id = resp.get(); + ids.add(id == null ? null : id.toString()); } synchronized (statsLock) { producedTotal += ids.size(); @@ -185,7 +184,7 @@ private static Map encodeFields(String eventType, Map */ public void ensureGroup(String group, String startId) { - try (Jedis jedis = pool.getResource()) { + try { jedis.xgroupCreate(streamKey, group, new StreamEntryID(startId), true); } catch (JedisDataException exc) { if (exc.getMessage() == null || !exc.getMessage().contains("BUSYGROUP")) { @@ -195,7 +194,7 @@ public void ensureGroup(String group, String startId) { } public long deleteGroup(String group) { - try (Jedis jedis = pool.getResource()) { + try { return jedis.xgroupDestroy(streamKey, group); } catch (JedisDataException exc) { return 0L; @@ -212,20 +211,17 @@ public long deleteGroup(String group) { * that recovery path).

*/ public List consume(String group, String consumer, int count, long blockMs) { - try (Jedis jedis = pool.getResource()) { - XReadGroupParams params = new XReadGroupParams() - .count(count) - .block((int) blockMs); - Map streams = new LinkedHashMap<>(); - // ``UNRECEIVED_ENTRY`` is the Jedis sentinel that serialises - // to the special ID ``>``: "deliver entries this group has - // not yet delivered to anyone". Same field name across 5.x - // and 6.x; 6.x also exposes it as ``XREADGROUP_UNDELIVERED_ENTRY``. - streams.put(streamKey, StreamEntryID.UNRECEIVED_ENTRY); - List>> result = - jedis.xreadGroup(group, consumer, params, streams); - return flattenEntries(result); - } + XReadGroupParams params = new XReadGroupParams() + .count(count) + .block((int) blockMs); + Map streams = new LinkedHashMap<>(); + // ``XREADGROUP_UNDELIVERED_ENTRY`` is the Jedis sentinel that + // serialises to the special ID ``>``: "deliver entries this group + // has not yet delivered to anyone". + streams.put(streamKey, StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY); + List>> result = + jedis.xreadGroup(group, consumer, params, streams); + return flattenEntries(result); } /** @@ -239,14 +235,12 @@ public List consume(String group, String consumer, int count, long blockM * consumer (or {@code XAUTOCLAIM}) handed to it.

*/ public List consumeOwnPel(String group, String consumer, int count) { - try (Jedis jedis = pool.getResource()) { - XReadGroupParams params = new XReadGroupParams().count(count); - Map streams = new LinkedHashMap<>(); - streams.put(streamKey, new StreamEntryID(0L, 0L)); - List>> result = - jedis.xreadGroup(group, consumer, params, streams); - return flattenEntries(result); - } + XReadGroupParams params = new XReadGroupParams().count(count); + Map streams = new LinkedHashMap<>(); + streams.put(streamKey, new StreamEntryID(0L, 0L)); + List>> result = + jedis.xreadGroup(group, consumer, params, streams); + return flattenEntries(result); } public long ack(String group, List ids) { @@ -258,9 +252,7 @@ public long ack(String group, List ids) { entryIds[i] = new StreamEntryID(ids.get(i)); } long acked; - try (Jedis jedis = pool.getResource()) { - acked = jedis.xack(streamKey, group, entryIds); - } + acked = jedis.xack(streamKey, group, entryIds); synchronized (statsLock) { ackedTotal += acked; } @@ -289,25 +281,23 @@ public AutoClaimResult autoclaim( List claimedAll = new ArrayList<>(); List deletedAll = new ArrayList<>(); String cursor = (startId == null || startId.isEmpty()) ? "0-0" : startId; - try (Jedis jedis = pool.getResource()) { - for (int page = 0; page < maxPages; page++) { - // Use sendCommand to get the raw 3-element reply - // (next-id, claimed-entries, deleted-ids). Jedis 6's - // typed xautoclaim wrapper hides the deleted-ids slot. - Object raw = jedis.sendCommand( - XAutoClaimRaw.XAUTOCLAIM, - streamKey, group, consumer, - Integer.toString(claimMinIdleMs), - cursor, - "COUNT", Integer.toString(pageCount)); - ParsedAutoClaim parsed = parseAutoClaim(raw); - claimedAll.addAll(parsed.claimed); - deletedAll.addAll(parsed.deletedIds); - if ("0-0".equals(parsed.nextCursor)) { - break; - } - cursor = parsed.nextCursor; + for (int page = 0; page < maxPages; page++) { + // Use executeCommand to get the raw 3-element reply + // (next-id, claimed-entries, deleted-ids). Jedis 6's + // typed xautoclaim wrapper hides the deleted-ids slot. + Object raw = jedis.executeCommand( + new CommandArguments(XAutoClaimRaw.XAUTOCLAIM).addObjects( + streamKey, group, consumer, + Integer.toString(claimMinIdleMs), + cursor, + "COUNT", Integer.toString(pageCount))); + ParsedAutoClaim parsed = parseAutoClaim(raw); + claimedAll.addAll(parsed.claimed); + deletedAll.addAll(parsed.deletedIds); + if ("0-0".equals(parsed.nextCursor)) { + break; } + cursor = parsed.nextCursor; } synchronized (statsLock) { claimedTotal += claimedAll.size(); @@ -404,7 +394,7 @@ private static String decodeString(Object raw) { * and is exposed only for explicit cleanup.

*/ public long deleteConsumer(String group, String consumer) { - try (Jedis jedis = pool.getResource()) { + try { return jedis.xgroupDelConsumer(streamKey, group, consumer); } catch (JedisDataException exc) { return 0L; @@ -427,27 +417,25 @@ public long deleteConsumer(String group, String consumer) { */ public int handoverPending(String group, String fromConsumer, String toConsumer, int batch) { int total = 0; - try (Jedis jedis = pool.getResource()) { - while (true) { - XPendingParams params = new XPendingParams() - .idle(0L) - .count(batch) - .consumer(fromConsumer); - List rows = jedis.xpending(streamKey, group, params); - if (rows == null || rows.isEmpty()) { - break; - } - StreamEntryID[] ids = new StreamEntryID[rows.size()]; - for (int i = 0; i < rows.size(); i++) { - ids[i] = rows.get(i).getID(); - } - XClaimParams claimParams = new XClaimParams(); - List claimed = jedis.xclaim( - streamKey, group, toConsumer, 0L, claimParams, ids); - total += (claimed == null ? 0 : claimed.size()); - if (rows.size() < batch) { - break; - } + while (true) { + XPendingParams params = new XPendingParams() + .idle(0L) + .count(batch) + .consumer(fromConsumer); + List rows = jedis.xpending(streamKey, group, params); + if (rows == null || rows.isEmpty()) { + break; + } + StreamEntryID[] ids = new StreamEntryID[rows.size()]; + for (int i = 0; i < rows.size(); i++) { + ids[i] = rows.get(i).getID(); + } + XClaimParams claimParams = new XClaimParams(); + List claimed = jedis.xclaim( + streamKey, group, toConsumer, 0L, claimParams, ids); + total += (claimed == null ? 0 : claimed.size()); + if (rows.size() < batch) { + break; } } synchronized (statsLock) { @@ -469,14 +457,12 @@ public int handoverPending(String group, String fromConsumer, String toConsumer, * stream.

*/ public List replay(String startId, String endId, int count) { - try (Jedis jedis = pool.getResource()) { - List rows = jedis.xrange(streamKey, startId, endId, count); - return toEntries(rows); - } + List rows = jedis.xrange(streamKey, startId, endId, count); + return toEntries(rows); } public long length() { - try (Jedis jedis = pool.getResource()) { + try { return jedis.xlen(streamKey); } catch (JedisDataException exc) { return 0L; @@ -484,7 +470,7 @@ public long length() { } public long trimMaxlen(long maxlen) { - try (Jedis jedis = pool.getResource()) { + try { XTrimParams params = new XTrimParams().maxLen(maxlen).approximateTrimming(); return jedis.xtrim(streamKey, params); } catch (JedisDataException exc) { @@ -493,7 +479,7 @@ public long trimMaxlen(long maxlen) { } public long trimMinid(String minid) { - try (Jedis jedis = pool.getResource()) { + try { XTrimParams params = new XTrimParams().minId(minid).approximateTrimming(); return jedis.xtrim(streamKey, params); } catch (JedisDataException exc) { @@ -512,7 +498,7 @@ public Map infoStream() { info.put("last_generated_id", null); info.put("first_entry_id", null); info.put("last_entry_id", null); - try (Jedis jedis = pool.getResource()) { + try { StreamInfo raw = jedis.xinfoStream(streamKey); if (raw == null) { return info; @@ -532,7 +518,7 @@ public Map infoStream() { public List> infoGroups() { List> out = new ArrayList<>(); - try (Jedis jedis = pool.getResource()) { + try { List rows = jedis.xinfoGroups(streamKey); if (rows == null) { return out; @@ -557,7 +543,7 @@ public List> infoGroups() { public List> infoConsumers(String group) { List> out = new ArrayList<>(); - try (Jedis jedis = pool.getResource()) { + try { List rows = jedis.xinfoConsumers2(streamKey, group); if (rows == null) { return out; @@ -578,7 +564,7 @@ public List> infoConsumers(String group) { /** Per-entry PEL view (id, consumer, idle, deliveries). */ public List> pendingDetail(String group, int count) { List> out = new ArrayList<>(); - try (Jedis jedis = pool.getResource()) { + try { XPendingParams params = new XPendingParams() .start(StreamEntryID.MINIMUM_ID) .end(StreamEntryID.MAXIMUM_ID) @@ -603,7 +589,7 @@ public List> pendingDetail(String group, int count) { /** Reverse-range tail read, used by the demo to render the most recent entries. */ public List tail(int count) { - try (Jedis jedis = pool.getResource()) { + try { List rows = jedis.xrevrange(streamKey, "+", "-", count); return toEntries(rows); } catch (JedisDataException exc) { @@ -635,9 +621,7 @@ public void resetStats() { /** Drop the stream key entirely. Used by the demo's reset path. */ public void deleteStream() { - try (Jedis jedis = pool.getResource()) { - jedis.del(streamKey); - } + jedis.del(streamKey); } // ------------------------------------------------------------------ diff --git a/content/develop/use-cases/streaming/java-jedis/_index.md b/content/develop/use-cases/streaming/java-jedis/_index.md index 707c1873d2..e5c2b0cdc8 100644 --- a/content/develop/use-cases/streaming/java-jedis/_index.md +++ b/content/develop/use-cases/streaming/java-jedis/_index.md @@ -53,12 +53,11 @@ The `EventStream` class wraps the stream operations ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/streaming/java-jedis/EventStream.java)): ```java -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.RedisClient; -JedisPool pool = new JedisPool(new JedisPoolConfig(), "localhost", 6379); +RedisClient jedis = RedisClient.create("localhost", 6379); EventStream stream = new EventStream( - pool, + jedis, "demo:events:orders", 2000, // approximate MAXLEN retention guardrail 5000); // XAUTOCLAIM idle threshold (ms) @@ -130,9 +129,8 @@ The ID is `{milliseconds}-{sequence}`, monotonically increasing within the strea ```java public List produceBatch(List>> events) { List ids = new ArrayList<>(events.size()); - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); - List> responses = new ArrayList<>(events.size()); + List> responses = new ArrayList<>(events.size()); + try (Pipeline pipe = jedis.pipelined()) { for (Map.Entry> event : events) { Map fields = encodeFields(event.getKey(), event.getValue()); XAddParams params = new XAddParams() @@ -141,10 +139,10 @@ public List produceBatch(List>> ev responses.add(pipe.xadd(streamKey, params, fields)); } pipe.sync(); - for (Response resp : responses) { - StreamEntryID id = resp.get(); - ids.add(id == null ? null : id.toString()); - } + } + for (Response resp : responses) { + StreamEntryID id = resp.get(); + ids.add(id == null ? null : id.toString()); } return ids; } @@ -160,20 +158,18 @@ Each consumer in a group runs the same `XREADGROUP` loop. The special ID `>` mea ```java public List consume(String group, String consumer, int count, long blockMs) { - try (Jedis jedis = pool.getResource()) { - XReadGroupParams params = new XReadGroupParams() - .count(count) - .block((int) blockMs); - Map streams = new LinkedHashMap<>(); - streams.put(streamKey, StreamEntryID.UNRECEIVED_ENTRY); - List>> result = - jedis.xreadGroup(group, consumer, params, streams); - return flattenEntries(result); - } + XReadGroupParams params = new XReadGroupParams() + .count(count) + .block((int) blockMs); + Map streams = new LinkedHashMap<>(); + streams.put(streamKey, StreamEntryID.XREADGROUP_UNDELIVERED_ENTRY); + List>> result = + jedis.xreadGroup(group, consumer, params, streams); + return flattenEntries(result); } ``` -`blockMs` makes the call efficient even when the stream is idle: the client parks on the server until either an entry arrives or the timeout expires, so consumers don't busy-loop. The Jedis instance is acquired from a `JedisPool` with try-with-resources, so the blocking call holds *its* connection for the duration of the block but other handlers and workers can grab their own connections in parallel. +`blockMs` makes the call efficient even when the stream is idle: the client parks on the server until either an entry arrives or the timeout expires, so consumers don't busy-loop. The blocking call borrows a connection from the `RedisClient` pool and holds *it* for the duration of the block, but other handlers and workers can grab their own connections in parallel. Reading with an explicit ID like `0-0` instead of `>` does something different — it replays entries already delivered to *this* consumer name (its private PEL). That is the canonical recovery path when the same consumer restarts: catch up on its own pending entries first, then resume reading new ones (`consumeOwnPel` exposes that read). @@ -190,9 +186,7 @@ public long ack(String group, List ids) { for (int i = 0; i < ids.size(); i++) { entryIds[i] = new StreamEntryID(ids.get(i)); } - try (Jedis jedis = pool.getResource()) { - return jedis.xack(streamKey, group, entryIds); - } + return jedis.xack(streamKey, group, entryIds); } ``` @@ -244,25 +238,23 @@ public AutoClaimResult autoclaim( List claimedAll = new ArrayList<>(); List deletedAll = new ArrayList<>(); String cursor = (startId == null || startId.isEmpty()) ? "0-0" : startId; - try (Jedis jedis = pool.getResource()) { - for (int page = 0; page < maxPages; page++) { - // Use sendCommand for the raw 3-element XAUTOCLAIM reply - // (next-id, claimed-entries, deleted-ids). Jedis 6's typed - // xautoclaim wrapper returns only the first two slots. - Object raw = jedis.sendCommand( - XAutoClaimRaw.XAUTOCLAIM, - streamKey, group, consumer, - Integer.toString(claimMinIdleMs), - cursor, - "COUNT", Integer.toString(pageCount)); - ParsedAutoClaim parsed = parseAutoClaim(raw); - claimedAll.addAll(parsed.claimed); - deletedAll.addAll(parsed.deletedIds); - if ("0-0".equals(parsed.nextCursor)) { - break; - } - cursor = parsed.nextCursor; + for (int page = 0; page < maxPages; page++) { + // Use executeCommand for the raw 3-element XAUTOCLAIM reply + // (next-id, claimed-entries, deleted-ids). Jedis 8's typed + // xautoclaim wrapper returns only the first two slots. + Object raw = jedis.executeCommand( + new CommandArguments(XAutoClaimRaw.XAUTOCLAIM).addObjects( + streamKey, group, consumer, + Integer.toString(claimMinIdleMs), + cursor, + "COUNT", Integer.toString(pageCount))); + ParsedAutoClaim parsed = parseAutoClaim(raw); + claimedAll.addAll(parsed.claimed); + deletedAll.addAll(parsed.deletedIds); + if ("0-0".equals(parsed.nextCursor)) { + break; } + cursor = parsed.nextCursor; } return new AutoClaimResult(claimedAll, deletedAll); } @@ -272,7 +264,7 @@ A single `XAUTOCLAIM` call scans up to `pageCount` PEL entries starting at `star The `deletedIds` list contains PEL entry IDs whose stream payload was already trimmed by the time the claim ran (typically because `MAXLEN ~` retention outran a slow consumer). `XAUTOCLAIM` removes those dangling slots from the PEL itself, so the caller does *not* need to `XACK` them — but the entries cannot be retried either, so log and route them to a dead-letter store for offline inspection. Redis 7.0 introduced this third return element; the example requires Redis 7.0+ for that reason. -Jedis 6.x's typed `jedis.xautoclaim(...)` overload returns only the cursor and the claimed entries (as `Map.Entry>`) — the deleted-IDs slot is dropped. To surface it, the helper sends `XAUTOCLAIM` through `Jedis.sendCommand` and parses all three reply elements by hand. +Jedis 8.x's typed `jedis.xautoclaim(...)` overload returns only the cursor and the claimed entries (as `Map.Entry>`) — the deleted-IDs slot is dropped. To surface it, the helper sends `XAUTOCLAIM` through `RedisClient.executeCommand` and parses all three reply elements by hand. `reapIdlePel` is the right primitive for the recovery path because it claims and processes in one step: every entry the call returned is now in *this* consumer's PEL, so the same consumer is responsible for processing and acking it. In production each consumer thread runs `reapIdlePel` periodically (every few seconds, on a timer) so a crashed peer's entries never sit invisibly. The demo exposes it as a manual button so you can trigger the reap after waiting for the idle threshold. @@ -284,10 +276,8 @@ Jedis 6.x's typed `jedis.xautoclaim(...)` overload returns only the cursor and t ```java public List replay(String startId, String endId, int count) { - try (Jedis jedis = pool.getResource()) { - List rows = jedis.xrange(streamKey, startId, endId, count); - return toEntries(rows); - } + List rows = jedis.xrange(streamKey, startId, endId, count); + return toEntries(rows); } ``` @@ -340,7 +330,7 @@ Before running the demo, make sure that: * Redis 7.0 or later is running and accessible. By default, the demo connects to `localhost:6379`. `XAUTOCLAIM` was added in Redis 6.2, but its reply gained a third element (the list of deleted IDs) in 7.0; the example relies on that shape. * JDK 17 or later is installed (the demo's inline HTML uses text blocks, introduced as a preview in JDK 13 and finalised in JDK 15; 17+ keeps the demo on a current LTS). -* The Jedis JAR (5.0+; the example was tested against 6.2.0) and its dependencies are on your classpath. Get them from [Maven Central](https://repo1.maven.org/maven2/redis/clients/jedis/), or via Maven/Gradle in a project setup. You also need [`slf4j-api`](https://repo1.maven.org/maven2/org/slf4j/slf4j-api/) and [`commons-pool2`](https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/), which Jedis declares as transitive dependencies. +* The Jedis JAR (the example was tested against 8.0.2) and its dependencies are on your classpath. Get them from [Maven Central](https://repo1.maven.org/maven2/redis/clients/jedis/), or via Maven/Gradle in a project setup. You also need [`slf4j-api`](https://repo1.maven.org/maven2/org/slf4j/slf4j-api/), [`commons-pool2`](https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/), [`gson`](https://repo1.maven.org/maven2/com/google/code/gson/gson/), [`json`](https://repo1.maven.org/maven2/org/json/json/), and [`redis-authx-core`](https://repo1.maven.org/maven2/redis/clients/authentication/redis-authx-core/), which Jedis declares as transitive dependencies. If your Redis server is running elsewhere, start the demo with `--redis-host` and `--redis-port`. @@ -358,15 +348,21 @@ curl -O $BASE/ConsumerWorker.java curl -O $BASE/DemoServer.java ``` -Download the JARs into the same directory (the example assumes Jedis 6.2.0, `slf4j-api` 2.0.12, and `commons-pool2` 2.12.1 — adjust the filenames if you use different versions): +Download the JARs into the same directory (the example assumes Jedis 8.0.2, `slf4j-api` 1.7.36, `commons-pool2` 2.13.1, `gson` 2.14.0, `json` 20260719, and `redis-authx-core` 0.1.1-beta2 — adjust the filenames if you use different versions): ```bash -JEDIS_VER=6.2.0 -SLF4J_VER=2.0.12 -POOL_VER=2.12.1 +JEDIS_VER=8.0.2 +SLF4J_VER=1.7.36 +POOL_VER=2.13.1 +GSON_VER=2.14.0 +JSON_VER=20260719 +AUTHX_VER=0.1.1-beta2 curl -O https://repo1.maven.org/maven2/redis/clients/jedis/$JEDIS_VER/jedis-$JEDIS_VER.jar curl -O https://repo1.maven.org/maven2/org/slf4j/slf4j-api/$SLF4J_VER/slf4j-api-$SLF4J_VER.jar curl -O https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/$POOL_VER/commons-pool2-$POOL_VER.jar +curl -O https://repo1.maven.org/maven2/com/google/code/gson/gson/$GSON_VER/gson-$GSON_VER.jar +curl -O https://repo1.maven.org/maven2/org/json/json/$JSON_VER/json-$JSON_VER.jar +curl -O https://repo1.maven.org/maven2/redis/clients/authentication/redis-authx-core/$AUTHX_VER/redis-authx-core-$AUTHX_VER.jar ``` ### Start the demo server @@ -374,10 +370,10 @@ curl -O https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/$POOL_VE From that directory: ```bash -javac -cp jedis-6.2.0.jar:slf4j-api-2.0.12.jar:commons-pool2-2.12.1.jar \ +javac -cp jedis-8.0.2.jar:slf4j-api-1.7.36.jar:commons-pool2-2.13.1.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar \ EventStream.java ConsumerWorker.java DemoServer.java -java -cp .:jedis-6.2.0.jar:slf4j-api-2.0.12.jar:commons-pool2-2.12.1.jar \ +java -cp .:jedis-8.0.2.jar:slf4j-api-1.7.36.jar:commons-pool2-2.13.1.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar \ DemoServer --port 8083 --redis-host localhost --redis-port 6379 ``` @@ -442,11 +438,11 @@ A single Redis Stream is a single key, and on a Redis Cluster a single key lives Per-entity partitioning (`events:order:{order_id}`) is the canonical pattern when you treat each entity's stream as the event-sourcing log for that entity: every state change for one order goes on its own stream, which is also bounded in size by the entity's lifetime. -### Tune the JedisPool +### Tune the connection pool -The demo bumps `JedisPoolConfig.setMaxTotal(64)` and `setMaxIdle(32)` because every blocking `XREADGROUP` call holds *its* Jedis instance for the full block duration (here 500 ms by default). If the pool were sized at the default 8, three consumers blocking at once would leave only five connections for the 16 HTTP handlers, which can starve `/state` polling under load. In production size the pool to at least *(blocking consumers) + (concurrent request threads) + headroom* and keep that ceiling below the server-side `maxclients`. +The demo builds the client with a `ConnectionPoolConfig` and calls `setMaxTotal(64)` and `setMaxIdle(32)` because every blocking `XREADGROUP` call holds *its* pooled connection for the full block duration (here 500 ms by default). If the pool were sized at the default 8, three consumers blocking at once would leave only five connections for the 16 HTTP handlers, which can starve `/state` polling under load. In production size the pool to at least *(blocking consumers) + (concurrent request threads) + headroom* and keep that ceiling below the server-side `maxclients`. -Every helper acquires its Jedis from the pool with try-with-resources, so connections always go back to the pool on success or exception. No in-process lock is needed across helpers because each call uses its own connection. +Every helper call borrows a connection from the pool and returns it when the call finishes, and pipelines return theirs when try-with-resources closes them, so connections always go back to the pool on success or exception. No in-process lock is needed across helpers because each call uses its own connection. ### Don't read with XREAD (no group) and then try to ack diff --git a/content/develop/use-cases/time-series-dashboard/java-jedis/DemoServer.java b/content/develop/use-cases/time-series-dashboard/java-jedis/DemoServer.java index ef43559d28..c18283dac4 100644 --- a/content/develop/use-cases/time-series-dashboard/java-jedis/DemoServer.java +++ b/content/develop/use-cases/time-series-dashboard/java-jedis/DemoServer.java @@ -14,9 +14,8 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import redis.clients.jedis.DefaultJedisClientConfig; import redis.clients.jedis.HostAndPort; -import redis.clients.jedis.JedisPool; +import redis.clients.jedis.RedisClient; import redis.clients.jedis.exceptions.JedisDataException; /** @@ -37,12 +36,9 @@ private DemoServer() { public static void main(String[] args) throws Exception { Arguments parsed = Arguments.parse(args); - JedisPool jedisPool = new JedisPool( - new HostAndPort(parsed.redisHost(), parsed.redisPort()), - DefaultJedisClientConfig.builder().build() - ); + RedisClient jedis = RedisClient.create(new HostAndPort(parsed.redisHost(), parsed.redisPort())); - RedisTimeSeriesStore store = new RedisTimeSeriesStore(jedisPool, SensorSimulator.SENSORS); + RedisTimeSeriesStore store = new RedisTimeSeriesStore(jedis, SensorSimulator.SENSORS); try { store.ensureSchema(); } catch (JedisDataException exception) { @@ -88,7 +84,7 @@ public static void main(String[] args) throws Exception { Runtime.getRuntime().addShutdownHook(new Thread(() -> { ingestExecutor.shutdownNow(); server.stop(0); - jedisPool.close(); + jedis.close(); })); System.out.println("Dashboard running at http://localhost:" + parsed.port()); diff --git a/content/develop/use-cases/time-series-dashboard/java-jedis/RedisTimeSeriesStore.java b/content/develop/use-cases/time-series-dashboard/java-jedis/RedisTimeSeriesStore.java index c7b46b8151..b05779d9b4 100644 --- a/content/develop/use-cases/time-series-dashboard/java-jedis/RedisTimeSeriesStore.java +++ b/content/develop/use-cases/time-series-dashboard/java-jedis/RedisTimeSeriesStore.java @@ -4,9 +4,8 @@ import java.util.List; import java.util.Map; -import redis.clients.jedis.Connection; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisPool; +import redis.clients.jedis.CommandArguments; +import redis.clients.jedis.RedisClient; import redis.clients.jedis.commands.ProtocolCommand; import redis.clients.jedis.exceptions.JedisDataException; @@ -19,11 +18,11 @@ public final class RedisTimeSeriesStore { public static final int BUCKET_MS = 3_000; public static final int RETENTION_MS = 12_000; - private final JedisPool jedisPool; + private final RedisClient jedis; private final List sensors; - public RedisTimeSeriesStore(JedisPool jedisPool, List sensors) { - this.jedisPool = jedisPool; + public RedisTimeSeriesStore(RedisClient jedis, List sensors) { + this.jedis = jedis; this.sensors = sensors; } @@ -39,8 +38,8 @@ public void ensureSchema() { args.add(value); }); - try (Jedis jedis = jedisPool.getResource()) { - executeCommand(jedis.getConnection(), TimeSeriesCommand.TS_CREATE, args.toArray(String[]::new)); + try { + executeCommand(TimeSeriesCommand.TS_CREATE, args.toArray(String[]::new)); } catch (JedisDataException exception) { if (!exception.getMessage().toLowerCase().contains("key already exists")) { throw exception; @@ -62,9 +61,7 @@ public void addSamples(List samples) { args.add(String.valueOf(sample.value())); } - try (Jedis jedis = jedisPool.getResource()) { - executeCommand(jedis.getConnection(), TimeSeriesCommand.TS_MADD, args.toArray(String[]::new)); - } + executeCommand(TimeSeriesCommand.TS_MADD, args.toArray(String[]::new)); } public Map dashboardSnapshot() { @@ -122,16 +119,13 @@ public Map dashboardSnapshot() { } private List> range(SensorSimulator.SensorDefinition sensor, long startMs, long endMs) { - try (Jedis jedis = jedisPool.getResource()) { - Object response = executeCommand( - jedis.getConnection(), - TimeSeriesCommand.TS_RANGE, - sensor.key(), - String.valueOf(startMs), - String.valueOf(endMs) - ); - return parsePoints(response); - } + Object response = executeCommand( + TimeSeriesCommand.TS_RANGE, + sensor.key(), + String.valueOf(startMs), + String.valueOf(endMs) + ); + return parsePoints(response); } private List> aggregate( @@ -139,26 +133,23 @@ private List> aggregate( long startMs, long endMs, String aggregation) { - try (Jedis jedis = jedisPool.getResource()) { - Object response = executeCommand( - jedis.getConnection(), - TimeSeriesCommand.TS_RANGE, - sensor.key(), - String.valueOf(startMs), - String.valueOf(endMs), - "ALIGN", - "0", - "AGGREGATION", - aggregation, - String.valueOf(BUCKET_MS) - ); - return parsePoints(response); - } + Object response = executeCommand( + TimeSeriesCommand.TS_RANGE, + sensor.key(), + String.valueOf(startMs), + String.valueOf(endMs), + "ALIGN", + "0", + "AGGREGATION", + aggregation, + String.valueOf(BUCKET_MS) + ); + return parsePoints(response); } private Map latest(SensorSimulator.SensorDefinition sensor) { - try (Jedis jedis = jedisPool.getResource()) { - Object response = executeCommand(jedis.getConnection(), TimeSeriesCommand.TS_GET, sensor.key()); + try { + Object response = executeCommand(TimeSeriesCommand.TS_GET, sensor.key()); if (!(response instanceof List values) || values.size() < 2) { return null; } @@ -202,9 +193,8 @@ private List> parsePoints(Object response) { return parsed; } - private Object executeCommand(Connection connection, ProtocolCommand command, String... args) { - connection.sendCommand(command, args); - return connection.getOne(); + private Object executeCommand(ProtocolCommand command, String... args) { + return jedis.executeCommand(new CommandArguments(command).addObjects((Object[]) args)); } private long asLong(Object value) { diff --git a/content/develop/use-cases/time-series-dashboard/java-jedis/_index.md b/content/develop/use-cases/time-series-dashboard/java-jedis/_index.md index 5d1c8c2b9d..d63da98e3b 100644 --- a/content/develop/use-cases/time-series-dashboard/java-jedis/_index.md +++ b/content/develop/use-cases/time-series-dashboard/java-jedis/_index.md @@ -101,7 +101,7 @@ If you use Maven: redis.clients jedis - 5.0.1 + 8.0.2 ``` @@ -124,14 +124,14 @@ curl -O $BASE/DemoServer.java Compile and run the demo: ```bash -javac -cp jedis-5.0.1.jar:commons-pool2-2.12.1.jar SensorSimulator.java RedisTimeSeriesStore.java DemoServer.java -java -cp .:jedis-5.0.1.jar:commons-pool2-2.12.1.jar DemoServer +javac -cp jedis-8.0.2.jar:commons-pool2-2.13.1.jar:slf4j-api-1.7.36.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar SensorSimulator.java RedisTimeSeriesStore.java DemoServer.java +java -cp .:jedis-8.0.2.jar:commons-pool2-2.13.1.jar:slf4j-api-1.7.36.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar DemoServer ``` The server accepts optional flags if your Redis instance is not on the default host and port: ```bash -java -cp .:jedis-5.0.1.jar:commons-pool2-2.12.1.jar DemoServer --redis-host 127.0.0.1 --redis-port 6379 --port 8080 +java -cp .:jedis-8.0.2.jar:commons-pool2-2.13.1.jar:slf4j-api-1.7.36.jar:gson-2.14.0.jar:json-20260719.jar:redis-authx-core-0.1.1-beta2.jar DemoServer --redis-host 127.0.0.1 --redis-port 6379 --port 8080 ``` After starting the server, visit `http://localhost:8080`. From 3ef1c870941a05ce10f56bf598638a2583b3a533 Mon Sep 17 00:00:00 2001 From: Andy Stark Date: Fri, 30 Oct 2026 13:53:53 +0000 Subject: [PATCH 2/2] DOC-7226 Return a pub-sub subscription's connection even if UNSUBSCRIBE never completes Cursor Bugbot found that Subscription.close() could no longer force a stuck subscribe() to return. The earlier change in this PR called jedis.subscribe(listener, ...) on the shared RedisClient, which borrows its connection internally, so close() had no handle on it. On a half-open socket UNSUBSCRIBE never completes and the pooled connection stays checked out for the life of the process. Each Subscription now borrows its own Connection with jedis.getPool().getResource() and runs listener.proceed(connection, ...) or proceedWithPatterns(...) on it, which is what UnifiedJedis.subscribe does internally. The worker thread returns the connection in a finally block, and close() restores the last resort the pre-migration code had: if the thread has not exited a second after UNSUBSCRIBE, it disconnects the connection so the blocked read fails. Reproduced first, then fixed. A TCP proxy between the demo and Redis was frozen to make a half-open link: before, close() returned after about 1 s and the pool still showed 1 active connection; after, close() takes about 1.5 s and the pool shows 0. The normal path is unchanged: delivery counts exact (203 = 3 + 200), PUBSUB introspection correct, 150 subscribe/unsubscribe cycles against a pool of 64, no deprecation warnings, nested-borrow check clean. The page text that described subscribe() borrowing the connection now describes the borrow. Constraint: UnifiedJedis.subscribe() borrows its connection internally; a caller that needs to break a stuck subscription must borrow the Connection itself and call JedisPubSub.proceed(connection, ...) Learned: to reproduce a stuck unsubscribe, freeze a TCP proxy between the client and Redis rather than killing either end; a dead socket makes subscribe() throw and masks the bug Ticket: DOC-7226 Co-Authored-By: Claude Sonnet 5.5 --- .../pub-sub/java-jedis/RedisPubSubHub.java | 44 +++++++++++++++---- .../use-cases/pub-sub/java-jedis/_index.md | 33 ++++++++------ 2 files changed, 56 insertions(+), 21 deletions(-) diff --git a/content/develop/use-cases/pub-sub/java-jedis/RedisPubSubHub.java b/content/develop/use-cases/pub-sub/java-jedis/RedisPubSubHub.java index 5002b09ea5..f1f2f879dc 100644 --- a/content/develop/use-cases/pub-sub/java-jedis/RedisPubSubHub.java +++ b/content/develop/use-cases/pub-sub/java-jedis/RedisPubSubHub.java @@ -1,6 +1,7 @@ import redis.clients.jedis.BuilderFactory; import redis.clients.jedis.CommandArguments; import redis.clients.jedis.CommandObject; +import redis.clients.jedis.Connection; import redis.clients.jedis.JedisPubSub; import redis.clients.jedis.Protocol; import redis.clients.jedis.RedisClient; @@ -74,7 +75,7 @@ public Map toMap() { * *

Each {@code Subscription} owns its own {@link JedisPubSub} listener * running on a dedicated thread with a dedicated pooled - * connection. Subscriptions are independent: closing one does not + * {@link Connection}. Subscriptions are independent: closing one does not * affect another, even if they share channels.

*/ public final class Subscription { @@ -88,6 +89,7 @@ public final class Subscription { private long received = 0; private volatile boolean closed = false; + private final Connection connection; private final JedisPubSub listener; private final Thread thread; @@ -100,10 +102,13 @@ public final class Subscription { this.isPattern = isPattern; this.bufferSize = bufferSize; - // Each subscribe() call borrows its own pooled connection for as - // long as the subscription lasts. Sharing one connection across - // subscribers would couple their lifetimes — unsubscribing one - // would close the channel for the others. + // Each subscriber gets its own pooled connection. Sharing one + // connection across subscribers would couple their lifetimes — + // unsubscribing one would close the channel for the others. The + // helper borrows the connection itself, instead of calling + // jedis.subscribe(), so that close() can break it if UNSUBSCRIBE + // never completes. + this.connection = jedis.getPool().getResource(); // ackLatch is counted down by Jedis once per target after Redis // acknowledges the SUBSCRIBE / PSUBSCRIBE. The constructor blocks @@ -138,17 +143,25 @@ public void onPSubscribe(String pattern, int subscribedChannels) { this.thread = new Thread(() -> { try { if (isPattern) { - jedis.psubscribe(listener, targetArr); + listener.proceedWithPatterns(connection, targetArr); } else { - jedis.subscribe(listener, targetArr); + listener.proceed(connection, targetArr); } } catch (Exception e) { + // Connection closed by close() is expected on shutdown. // If we never got the first ack through, surface the // failure so the constructor's wait can give up. startupError.compareAndSet(0L, 1L); while (ackLatch.getCount() > 0) { ackLatch.countDown(); } + } finally { + // Return the connection to the pool (a broken one is + // discarded rather than reused). + try { + connection.close(); + } catch (Exception ignored) { + } } }, "pubsub-" + name); this.thread.setDaemon(true); @@ -231,7 +244,9 @@ public void resetReceived() { * the same socket; after the listener has unsubscribed from every * target the blocking {@code subscribe}/{@code psubscribe} call * returns, the thread exits, and the connection goes back to the - * pool.

+ * pool. If that does not happen within a second (for example on a + * half-open socket) the connection is disconnected, which makes the + * blocked read fail so the thread can return it.

*/ public void close() { if (closed) { @@ -253,6 +268,19 @@ public void close() { } catch (InterruptedException e) { Thread.currentThread().interrupt(); } + if (thread.isAlive()) { + // Last resort: dropping the connection forces the + // blocking proceed() call to return. + try { + connection.disconnect(); + } catch (Exception ignored) { + } + try { + thread.join(500); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } } public boolean isAlive() { diff --git a/content/develop/use-cases/pub-sub/java-jedis/_index.md b/content/develop/use-cases/pub-sub/java-jedis/_index.md index bcad397c45..9f6793e16a 100644 --- a/content/develop/use-cases/pub-sub/java-jedis/_index.md +++ b/content/develop/use-cases/pub-sub/java-jedis/_index.md @@ -36,7 +36,7 @@ In this example, the publisher side calls `PUBLISH` with a JSON-encoded body and The flow looks like this: 1. The application calls `hub.subscribe(name, channels)` or `hub.psubscribe(name, patterns)` -2. The helper builds a `JedisPubSub` listener and starts a daemon thread that calls `jedis.subscribe(listener, channels...)` (or `psubscribe(...)`) on the shared `RedisClient` — each call borrows a dedicated connection from the pool and blocks for the lifetime of the subscription +2. The helper borrows a dedicated connection from the `RedisClient` pool, builds a `JedisPubSub` listener, and starts a daemon thread that calls `listener.proceed(connection, channels...)` (or `proceedWithPatterns(...)`) — both calls block for the lifetime of the subscription 3. The application (or another process) calls `hub.publish(channel, message)` 4. Redis fans the message out over every subscribing client's open socket 5. The listener's `onMessage` (or `onPMessage`) callback fires on the subscriber thread, wraps the raw message as a `ReceivedMessage`, and appends it to a per-subscriber ring buffer @@ -100,8 +100,9 @@ Subscription (in-process, one per subscriber) isPattern boolean buffer Deque (capped, default 50) received long - listener JedisPubSub (subscribed on its own connection) - thread Thread (blocks inside jedis.subscribe()) + connection Connection (borrowed from the pool, one per subscription) + listener JedisPubSub (subscribed on that connection) + thread Thread (blocks inside listener.proceed()) ``` The implementation uses: @@ -112,7 +113,7 @@ The implementation uses: * [`PUBSUB CHANNELS`](/content/commands/pubsub-channels.md) to list the channels with at least one active exact-match subscriber * [`PUBSUB NUMSUB`](/content/commands/pubsub-numsub.md) to count subscribers per channel * [`PUBSUB NUMPAT`](/content/commands/pubsub-numpat.md) to count active pattern subscriptions server-wide -* Jedis' `JedisPubSub` listener subclass plus a dedicated daemon thread per subscriber so the blocking `subscribe()`/`psubscribe()` call does not freeze the rest of the application +* Jedis' `JedisPubSub` listener subclass plus a dedicated daemon thread per subscriber so the blocking `proceed()`/`proceedWithPatterns()` call does not freeze the rest of the application ## Publishing messages @@ -144,6 +145,8 @@ public Subscription subscribe(String name, List channels) { Inside the `Subscription` constructor, the helper builds a `JedisPubSub` listener, starts the dispatch thread, and blocks until Redis has acknowledged every `SUBSCRIBE` so the very first publish can't race ahead: ```java +this.connection = jedis.getPool().getResource(); + final CountDownLatch ackLatch = new CountDownLatch(targets.size()); this.listener = new JedisPubSub() { @@ -159,10 +162,14 @@ this.listener = new JedisPubSub() { final String[] targetArr = this.targets.toArray(new String[0]); this.thread = new Thread(() -> { - if (isPattern) { - jedis.psubscribe(listener, targetArr); - } else { - jedis.subscribe(listener, targetArr); + try { + if (isPattern) { + listener.proceedWithPatterns(connection, targetArr); + } else { + listener.proceed(connection, targetArr); + } + } finally { + connection.close(); // back to the pool } }, "pubsub-" + name); this.thread.setDaemon(true); @@ -175,8 +182,8 @@ if (!ackLatch.await(2, TimeUnit.SECONDS)) { A few details matter here: -* `jedis.subscribe(...)` and `jedis.psubscribe(...)` **block the calling thread** until the listener unsubscribes from every target. Running each subscription on its own thread is the only way to keep the rest of the application responsive — there is no "run in background" helper in Jedis. -* Each `subscribe()` or `psubscribe()` call borrows its own connection from the pool and keeps it until the subscription ends. A connection that has issued `SUBSCRIBE` or `PSUBSCRIBE` is switched into subscribe-only mode, so other commands on the same connection would fail — `RedisClient` runs those commands on other pooled connections. Sharing one connection across subscribers would also couple their lifetimes — unsubscribing one would tear down the channel for the others. +* `listener.proceed(...)` and `listener.proceedWithPatterns(...)` **block the calling thread** until the listener unsubscribes from every target. Running each subscription on its own thread is the only way to keep the rest of the application responsive — there is no "run in background" helper in Jedis. `jedis.subscribe(...)` blocks in the same way, but it borrows the connection internally, so the caller has no handle to break it if the unsubscribe never completes. +* Each `Subscription` borrows its own connection from the pool with `jedis.getPool().getResource()` and keeps it until the subscription ends. A connection that has issued `SUBSCRIBE` or `PSUBSCRIBE` is switched into subscribe-only mode, so other commands on the same connection would fail — `RedisClient` runs those commands on other pooled connections. Sharing one connection across subscribers would also couple their lifetimes — unsubscribing one would tear down the channel for the others. * The constructor waits on a `CountDownLatch` that's decremented once per target inside `onSubscribe` / `onPSubscribe`. Without it, the parent thread could return from `subscribe()` while the spawned worker thread is still about to send the SUBSCRIBE command — and a `PUBLISH` issued immediately afterwards would beat the subscription into Redis. * The `JedisPubSub` listener is the same object on both threads: it's safe to call `listener.unsubscribe()` from a different thread because Jedis ships the unsubscribe command back over the listener's own socket. After the unsubscribe round-trip completes, the blocking `subscribe()` call returns and the worker thread exits. @@ -323,15 +330,15 @@ A subscriber that's offline when a message is published misses it permanently. F ### Give every subscriber its own connection -A connection that has issued `SUBSCRIBE` or `PSUBSCRIBE` is in subscribe-only mode: any other command on the same connection will fail. Each `subscribe(...)` or `psubscribe(...)` call on a `RedisClient` borrows a connection from the pool and dedicates it to that subscription until it's torn down. The helper does this for you — every `Subscription` makes its own call on its own thread and holds its connection for the subscription's lifetime — but if you write your own subscriber code, the same rule applies. +A connection that has issued `SUBSCRIBE` or `PSUBSCRIBE` is in subscribe-only mode: any other command on the same connection will fail. Each `Subscription` borrows a connection from the `RedisClient` pool with `jedis.getPool().getResource()` and dedicates it to that subscription until it's torn down. The helper does this for you — every `Subscription` runs its own listener on its own thread and holds its connection for the subscription's lifetime — but if you write your own subscriber code, the same rule applies. A related corollary: never let the pool size cap your subscriber count. If you have 100 long-lived subscribers in one process, `ConnectionPoolConfig.setMaxTotal(...)` needs to allow at least 100 connections for the subscribers plus whatever your publishers and bookkeeping calls need. Set it generously; idle subscriber connections cost almost nothing on the Redis side. ### Unsubscribe through the listener, not the connection -`jedis.subscribe(listener, channels...)` blocks until the listener calls `listener.unsubscribe()` (or `listener.punsubscribe()` for a pattern subscription). It's safe to call those methods from any thread: Jedis ships the unsubscribe command back over the listener's socket, Redis acknowledges, and then the blocking `subscribe()` call returns and the worker thread exits. +`listener.proceed(connection, channels...)` blocks until the listener calls `listener.unsubscribe()` (or `listener.punsubscribe()` for a pattern subscription). It's safe to call those methods from any thread: Jedis ships the unsubscribe command back over the listener's socket, Redis acknowledges, and then the blocking `subscribe()` call returns and the worker thread exits. -Closing the underlying `Jedis` instance directly works as a fallback — the socket read errors out and the blocking call returns — but it's a less clean shutdown and produces a stack trace in the logs. The helper does both: it calls `listener.unsubscribe()` first, then closes the connection from the worker thread's `finally` block as a belt-and-braces guard. +Disconnecting the underlying `Connection` directly works as a fallback — the socket read errors out and the blocking call returns — but it's a less clean shutdown and produces a stack trace in the logs. The helper does both: it calls `listener.unsubscribe()` first and, if the worker thread has not exited within a second (for example on a half-open socket), disconnects the connection. The worker thread's `finally` block then returns it to the pool. ### Choose a topic naming convention up front