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}
- * connection. Subscriptions are independent: closing one does not
+ * running on a dedicated thread with a dedicated pooled
+ * {@link Connection}. Subscriptions are independent: closing one does not
* affect another, even if they share channels.
*/
public final class Subscription {
@@ -85,7 +89,7 @@ public final class Subscription {
private long received = 0;
private volatile boolean closed = false;
- private final Jedis jedis;
+ private final Connection connection;
private final JedisPubSub listener;
private final Thread thread;
@@ -98,10 +102,13 @@ public final class Subscription {
this.isPattern = isPattern;
this.bufferSize = bufferSize;
- // Each subscriber gets its own Jedis connection. Sharing one
+ // 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.
- this.jedis = pool.getResource();
+ // 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
@@ -136,9 +143,9 @@ 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.
@@ -149,8 +156,10 @@ public void onPSubscribe(String pattern, int subscribedChannels) {
ackLatch.countDown();
}
} finally {
+ // Return the connection to the pool (a broken one is
+ // discarded rather than reused).
try {
- jedis.close();
+ connection.close();
} catch (Exception ignored) {
}
}
@@ -234,9 +243,10 @@ 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. 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) {
@@ -260,9 +270,9 @@ public void close() {
}
if (thread.isAlive()) {
// Last resort: dropping the connection forces the
- // blocking subscribe() call to return.
+ // blocking proceed() call to return.
try {
- jedis.close();
+ connection.disconnect();
} catch (Exception ignored) {
}
try {
@@ -288,7 +298,7 @@ public Map toMap() {
}
}
- private final JedisPool pool;
+ private final RedisClient jedis;
private final int bufferSize;
private final Map subscriptions = new ConcurrentHashMap<>();
@@ -298,12 +308,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 +328,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 +381,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 +401,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..9f6793e16a 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 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
@@ -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"));
@@ -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
@@ -121,10 +122,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 +134,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,7 +145,7 @@ 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();
+this.connection = jedis.getPool().getResource();
final CountDownLatch ackLatch = new CountDownLatch(targets.size());
@@ -164,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);
@@ -180,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 `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.
+* `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.
@@ -217,7 +219,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 +249,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 +257,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 +281,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,17 +328,17 @@ 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 `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, `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
-`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
@@ -349,7 +354,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`.