Skip to content

DOC-7226 Move prefetch-cache, time-series-dashboard, pub-sub and streaming Jedis use cases to RedisClient #4389

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
andy-stark-redis merged 2 commits into from
Oct 30, 2026
Merged

DOC-7226 Move prefetch-cache, time-series-dashboard, pub-sub and streaming Jedis use cases to RedisClient #4389

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3885120
DOC-7226 Move four more Jedis use cases from JedisPool to RedisClient
andy-stark-redis Oct 30, 2026
3ef1c87
DOC-7226 Return a pub-sub subscription's connection even if UNSUBSCRI…
andy-stark-redis Oct 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 9 additions & 10 deletions content/develop/use-cases/prefetch-cache/java-jedis/DemoServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -28,9 +27,9 @@
* without ever falling back to the primary on the read path.</p>
*
* <pre>{@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
* }</pre>
*/
Expand All @@ -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";
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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);
}
}
Expand Down
97 changes: 40 additions & 57 deletions content/develop/use-cases/prefetch-cache/java-jedis/PrefetchCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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;

Expand All @@ -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;
}
Expand Down Expand Up @@ -96,8 +95,7 @@ public Result(Map<String, String> record, boolean hit, double redisLatencyMs) {
*/
public int bulkLoad(Iterable<Map<String, String>> records) {
int loaded = 0;
try (Jedis jedis = pool.getResource()) {
Pipeline pipe = jedis.pipelined();
try (Pipeline pipe = jedis.pipelined()) {
for (Map<String, String> record : records) {
if (record == null) {
continue;
Expand All @@ -114,7 +112,7 @@ public int bulkLoad(Iterable<Map<String, String>> records) {
}
if (loaded > 0) {
pipe.sync();
}
}
}
synchronized (statsLock) {
prefetched += loaded;
Expand All @@ -133,9 +131,7 @@ public Result get(String entityId) {
String cacheKey = cacheKey(entityId);
long startedNs = System.nanoTime();
Map<String, String> 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()) {
Expand Down Expand Up @@ -185,17 +181,14 @@ public void applyChange(Map<String, Object> 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;
}
Expand All @@ -214,67 +207,57 @@ public void applyChange(Map<String, Object> 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. */
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<String> scan = jedis.scan(cursor, params);
cursor = scan.getCursor();
List<String> 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<String> scan = jedis.scan(cursor, params);
cursor = scan.getCursor();
List<String> 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;
}

/** Return every entity id currently in the cache, sorted. */
public List<String> ids() {
List<String> result = new ArrayList<>();
ScanParams params = new ScanParams().match(prefix + "*").count(500);
try (Jedis jedis = pool.getResource()) {
String cursor = ScanParams.SCAN_POINTER_START;
do {
ScanResult<String> 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<String> 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;
}

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<String> 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<String> 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. */
Expand Down
33 changes: 13 additions & 20 deletions content/develop/use-cases/prefetch-cache/java-jedis/_index.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down Expand Up @@ -95,8 +94,7 @@ The `bulkLoad` method pipelines a `DEL` + `HSET` + `EXPIRE` triple for every rec
```java
public int bulkLoad(Iterable<Map<String, String>> records) {
int loaded = 0;
try (Jedis jedis = pool.getResource()) {
Pipeline pipe = jedis.pipelined();
try (Pipeline pipe = jedis.pipelined()) {
for (Map<String, String> record : records) {
if (record == null) continue;
String entityId = record.get("id");
Expand All @@ -113,7 +111,7 @@ public int bulkLoad(Iterable<Map<String, String>> 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

Expand All @@ -124,9 +122,7 @@ public Result get(String entityId) {
String cacheKey = cacheKey(entityId);
long startedNs = System.nanoTime();
Map<String, String> 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()) {
Expand Down Expand Up @@ -158,17 +154,14 @@ public void applyChange(Map<String, Object> 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);
}
}
```
Expand Down Expand Up @@ -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
```

Expand Down Expand Up @@ -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

Expand Down
Loading
Loading