diff --git a/content/develop/use-cases/job-queue/java-jedis/DemoServer.java b/content/develop/use-cases/job-queue/java-jedis/DemoServer.java index 4301db7cf2..8ca60d4a05 100644 --- a/content/develop/use-cases/job-queue/java-jedis/DemoServer.java +++ b/content/develop/use-cases/job-queue/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; @@ -27,7 +27,7 @@ public class DemoServer { private static RedisJobQueue queue; private static WorkerPool pool; - private static JedisPool jedisPool; + private static RedisClient jedis; public static void main(String[] args) { String host = "127.0.0.1"; @@ -62,19 +62,22 @@ public static void main(String[] args) { } } - JedisPoolConfig poolConfig = new JedisPoolConfig(); + ConnectionPoolConfig poolConfig = new ConnectionPoolConfig(); poolConfig.setMaxTotal(32); 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); } - queue = new RedisJobQueue(jedisPool, "jobs-jedis", visibilityMs, 300, 50, 3); + queue = new RedisJobQueue(jedis, "jobs-jedis", visibilityMs, 300, 50, 3); pool = new WorkerPool(queue, 0, 400, 0.0, 0.0); if (initialWorkers > 0) { pool.resize(initialWorkers); @@ -106,7 +109,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/job-queue/java-jedis/RedisJobQueue.java b/content/develop/use-cases/job-queue/java-jedis/RedisJobQueue.java index 4fc2496db2..84e43d2881 100644 --- a/content/develop/use-cases/job-queue/java-jedis/RedisJobQueue.java +++ b/content/develop/use-cases/job-queue/java-jedis/RedisJobQueue.java @@ -1,6 +1,5 @@ -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisPool; import redis.clients.jedis.Pipeline; +import redis.clients.jedis.RedisClient; import redis.clients.jedis.args.ListDirection; import redis.clients.jedis.params.ScanParams; import redis.clients.jedis.resps.ScanResult; @@ -120,7 +119,7 @@ public class RedisJobQueue { private static final SecureRandom RANDOM = new SecureRandom(); - private final JedisPool pool; + private final RedisClient jedis; private final String queueName; private final long visibilityMs; private final int completedTtl; @@ -140,18 +139,18 @@ public class RedisJobQueue { private long failedTotal = 0; private long reclaimedTotal = 0; - public RedisJobQueue(JedisPool pool) { - this(pool, "jobs", 5000, 300, 50, 3); + public RedisJobQueue(RedisClient jedis) { + this(jedis, "jobs", 5000, 300, 50, 3); } public RedisJobQueue( - JedisPool pool, + RedisClient jedis, String queueName, long visibilityMs, int completedTtl, int completedHistory, int maxAttempts) { - this.pool = pool; + this.jedis = jedis; this.queueName = queueName; this.visibilityMs = visibilityMs; this.completedTtl = completedTtl; @@ -200,8 +199,7 @@ public String enqueue(Map payload) { meta.put("enqueued_at_ms", Long.toString(now)); meta.put("claim_token", ""); - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); + try (Pipeline pipe = jedis.pipelined()) { pipe.hset(metaKey(jobId), meta); pipe.lpush(pendingKey, jobId); pipe.sync(); @@ -219,9 +217,7 @@ public String enqueue(Map payload) { public ClaimedJob claim(long timeoutMs) { double timeoutSec = Math.max(timeoutMs / 1000.0, 0.1); String jobId; - try (Jedis jedis = pool.getResource()) { - jobId = jedis.blmove(pendingKey, processingKey, ListDirection.RIGHT, ListDirection.LEFT, timeoutSec); - } + jobId = jedis.blmove(pendingKey, processingKey, ListDirection.RIGHT, ListDirection.LEFT, timeoutSec); if (jobId == null) { return null; } @@ -230,18 +226,18 @@ public ClaimedJob claim(long timeoutMs) { long now = nowMs(); String mk = metaKey(jobId); Map meta; - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); + redis.clients.jedis.Response> resp; + try (Pipeline pipe = jedis.pipelined()) { Map updates = new LinkedHashMap<>(); updates.put("status", "processing"); updates.put("claimed_at_ms", Long.toString(now)); updates.put("claim_token", token); pipe.hset(mk, updates); pipe.hincrBy(mk, "attempts", 1); - redis.clients.jedis.Response> resp = pipe.hgetAll(mk); + resp = pipe.hgetAll(mk); pipe.sync(); - meta = resp.get(); } + meta = resp.get(); Map payload; try { @@ -274,18 +270,14 @@ public boolean complete(ClaimedJob job, Map result) { Integer.toString(completedHistory) ); Object res; - try (Jedis jedis = pool.getResource()) { - res = jedis.eval(COMPLETE_SCRIPT, keys, args); - } + res = jedis.eval(COMPLETE_SCRIPT, keys, args); if (res == null || !"1".equals(res.toString())) { return false; } - try (Jedis jedis = pool.getResource()) { - Map event = new LinkedHashMap<>(); - event.put("id", job.id); - event.put("status", "completed"); - jedis.publish(eventsChannel, JsonUtil.toJson(event)); - } + Map event = new LinkedHashMap<>(); + event.put("id", job.id); + event.put("status", "completed"); + jedis.publish(eventsChannel, JsonUtil.toJson(event)); synchronized (statsLock) { completedTotal++; } @@ -308,18 +300,14 @@ public boolean fail(ClaimedJob job, String error) { retry ? "1" : "0" ); Object res; - try (Jedis jedis = pool.getResource()) { - res = jedis.eval(FAIL_SCRIPT, keys, args); - } + res = jedis.eval(FAIL_SCRIPT, keys, args); if (res == null || "0".equals(res.toString())) { return false; } - try (Jedis jedis = pool.getResource()) { - Map event = new LinkedHashMap<>(); - event.put("id", job.id); - event.put("status", retry ? "retry" : "failed"); - jedis.publish(eventsChannel, JsonUtil.toJson(event)); - } + Map event = new LinkedHashMap<>(); + event.put("id", job.id); + event.put("status", retry ? "retry" : "failed"); + jedis.publish(eventsChannel, JsonUtil.toJson(event)); if (!retry) { synchronized (statsLock) { failedTotal++; @@ -334,9 +322,7 @@ public List reclaimStuck() { List keys = Arrays.asList(pendingKey, processingKey, metaPrefix); List args = Arrays.asList(Long.toString(nowMs()), Long.toString(visibilityMs)); Object res; - try (Jedis jedis = pool.getResource()) { - res = jedis.eval(RECLAIM_SCRIPT, keys, args); - } + res = jedis.eval(RECLAIM_SCRIPT, keys, args); List reclaimed = new ArrayList<>(); if (res instanceof List) { for (Object item : (List) res) { @@ -356,9 +342,7 @@ public List reclaimStuck() { /** Return the metadata hash for a job, or null if missing. */ public Map getJob(String jobId) { Map raw; - try (Jedis jedis = pool.getResource()) { - raw = jedis.hgetAll(metaKey(jobId)); - } + raw = jedis.hgetAll(metaKey(jobId)); if (raw == null || raw.isEmpty()) { return null; } @@ -388,30 +372,22 @@ public Map getJob(String jobId) { /** Return pending job IDs, oldest first. */ public List listPending() { List values; - try (Jedis jedis = pool.getResource()) { - values = jedis.lrange(pendingKey, 0, -1); - } + values = jedis.lrange(pendingKey, 0, -1); List reversed = new ArrayList<>(values); Collections.reverse(reversed); return reversed; } public List listProcessing() { - try (Jedis jedis = pool.getResource()) { - return jedis.lrange(processingKey, 0, -1); - } + return jedis.lrange(processingKey, 0, -1); } public List listCompleted() { - try (Jedis jedis = pool.getResource()) { - return jedis.lrange(completedKey, 0, -1); - } + return jedis.lrange(completedKey, 0, -1); } public List listFailed() { - try (Jedis jedis = pool.getResource()) { - return jedis.lrange(failedKey, 0, -1); - } + return jedis.lrange(failedKey, 0, -1); } /** Counters plus current queue depth. */ @@ -420,18 +396,21 @@ public Map stats() { long processing; long completed; long failed; - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); - redis.clients.jedis.Response pendingResp = pipe.llen(pendingKey); - redis.clients.jedis.Response processingResp = pipe.llen(processingKey); - redis.clients.jedis.Response completedResp = pipe.llen(completedKey); - redis.clients.jedis.Response failedResp = pipe.llen(failedKey); + redis.clients.jedis.Response pendingResp; + redis.clients.jedis.Response processingResp; + redis.clients.jedis.Response completedResp; + redis.clients.jedis.Response failedResp; + try (Pipeline pipe = jedis.pipelined()) { + pendingResp = pipe.llen(pendingKey); + processingResp = pipe.llen(processingKey); + completedResp = pipe.llen(completedKey); + failedResp = pipe.llen(failedKey); pipe.sync(); - pending = pendingResp.get(); - processing = processingResp.get(); - completed = completedResp.get(); - failed = failedResp.get(); } + pending = pendingResp.get(); + processing = processingResp.get(); + completed = completedResp.get(); + failed = failedResp.get(); Map out = new LinkedHashMap<>(); synchronized (statsLock) { out.put("enqueued_total", enqueuedTotal); @@ -458,19 +437,17 @@ public void resetStats() { /** Delete only queue:{name}:* keys via SCAN. */ public void purge() { - try (Jedis jedis = pool.getResource()) { - jedis.del(pendingKey, processingKey, completedKey, failedKey); - String cursor = ScanParams.SCAN_POINTER_START; - ScanParams params = new ScanParams().match(metaPrefix + "*").count(100); - do { - ScanResult result = jedis.scan(cursor, params); - List matched = result.getResult(); - if (!matched.isEmpty()) { - jedis.del(matched.toArray(new String[0])); - } - cursor = result.getCursor(); - } while (!cursor.equals("0")); - } + jedis.del(pendingKey, processingKey, completedKey, failedKey); + String cursor = ScanParams.SCAN_POINTER_START; + ScanParams params = new ScanParams().match(metaPrefix + "*").count(100); + do { + ScanResult result = jedis.scan(cursor, params); + List matched = result.getResult(); + if (!matched.isEmpty()) { + jedis.del(matched.toArray(new String[0])); + } + cursor = result.getCursor(); + } while (!cursor.equals("0")); resetStats(); } diff --git a/content/develop/use-cases/job-queue/java-jedis/_index.md b/content/develop/use-cases/job-queue/java-jedis/_index.md index b576518aa5..92ec1ac6ed 100644 --- a/content/develop/use-cases/job-queue/java-jedis/_index.md +++ b/content/develop/use-cases/job-queue/java-jedis/_index.md @@ -53,12 +53,12 @@ The `RedisJobQueue` class wraps the queue operations ([source](https://github.co ```java 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); - RedisJobQueue queue = new RedisJobQueue(pool, "jobs-jedis", 5000, 300, 50, 3); + RedisClient jedis = RedisClient.create("localhost", 6379); + RedisJobQueue queue = new RedisJobQueue(jedis, "jobs-jedis", 5000, 300, 50, 3); String jobId = queue.enqueue(Map.of( "kind", "email", @@ -82,7 +82,7 @@ public class Main { } ``` -Jedis operations are synchronous. The helper acquires a `Jedis` connection per call using `pool.getResource()` inside a try-with-resources block, so connections are returned to the pool even on errors. The blocking `claim()` method holds its own connection for the duration of the `BLMOVE` wait, which is fine because every other call uses a different connection. +Jedis operations are synchronous. The helper borrows a pooled connection for each call and returns it when the call finishes, and the pipeline calls return theirs when try-with-resources closes them, so connections are returned to the pool even on errors. The blocking `claim()` method holds its own connection for the duration of the `BLMOVE` wait, which is fine because every other call uses a different connection. ### Data model @@ -140,8 +140,7 @@ public String enqueue(Map payload) { meta.put("enqueued_at_ms", Long.toString(now)); meta.put("claim_token", ""); - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); + try (Pipeline pipe = jedis.pipelined()) { pipe.hset(metaKey(jobId), meta); pipe.lpush(pendingKey, jobId); pipe.sync(); @@ -160,9 +159,7 @@ A worker blocks until a job is available, then atomically pops it from the pendi public ClaimedJob claim(long timeoutMs) { double timeoutSec = Math.max(timeoutMs / 1000.0, 0.1); String jobId; - try (Jedis jedis = pool.getResource()) { - jobId = jedis.blmove(pendingKey, processingKey, ListDirection.RIGHT, ListDirection.LEFT, timeoutSec); - } + jobId = jedis.blmove(pendingKey, processingKey, ListDirection.RIGHT, ListDirection.LEFT, timeoutSec); if (jobId == null) { return null; } @@ -171,18 +168,18 @@ public ClaimedJob claim(long timeoutMs) { long now = System.currentTimeMillis(); String mk = metaKey(jobId); Map meta; - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); + Response> resp; + try (Pipeline pipe = jedis.pipelined()) { Map updates = new LinkedHashMap<>(); updates.put("status", "processing"); updates.put("claimed_at_ms", Long.toString(now)); updates.put("claim_token", token); pipe.hset(mk, updates); pipe.hincrBy(mk, "attempts", 1); - Response> resp = pipe.hgetAll(mk); + resp = pipe.hgetAll(mk); pipe.sync(); - meta = resp.get(); } + meta = resp.get(); // ... parse payload, attempts, and return a ClaimedJob ... } ``` @@ -206,9 +203,7 @@ public boolean complete(ClaimedJob job, Map result) { Integer.toString(completedHistory) ); Object res; - try (Jedis jedis = pool.getResource()) { - res = jedis.eval(COMPLETE_SCRIPT, keys, args); - } + res = jedis.eval(COMPLETE_SCRIPT, keys, args); if (res == null || !"1".equals(res.toString())) { return false; } @@ -237,9 +232,7 @@ public boolean fail(ClaimedJob job, String error) { retry ? "1" : "0" ); Object res; - try (Jedis jedis = pool.getResource()) { - res = jedis.eval(FAIL_SCRIPT, keys, args); - } + res = jedis.eval(FAIL_SCRIPT, keys, args); return res != null && !"0".equals(res.toString()); } ``` @@ -258,9 +251,7 @@ public List reclaimStuck() { Long.toString(visibilityMs) ); Object res; - try (Jedis jedis = pool.getResource()) { - res = jedis.eval(RECLAIM_SCRIPT, keys, args); - } + res = jedis.eval(RECLAIM_SCRIPT, keys, args); // ... unwrap the list of reclaimed IDs ... } ``` @@ -274,18 +265,18 @@ The Lua script also handles a narrower race: a worker that crashed between `BLMO ```java public Map stats() { long pending, processing, completed, failed; - try (Jedis jedis = pool.getResource()) { - Pipeline pipe = jedis.pipelined(); - Response pendingResp = pipe.llen(pendingKey); - Response processingResp = pipe.llen(processingKey); - Response completedResp = pipe.llen(completedKey); - Response failedResp = pipe.llen(failedKey); + Response pendingResp, processingResp, completedResp, failedResp; + try (Pipeline pipe = jedis.pipelined()) { + pendingResp = pipe.llen(pendingKey); + processingResp = pipe.llen(processingKey); + completedResp = pipe.llen(completedKey); + failedResp = pipe.llen(failedKey); pipe.sync(); - pending = pendingResp.get(); - processing = processingResp.get(); - completed = completedResp.get(); - failed = failedResp.get(); } + pending = pendingResp.get(); + processing = processingResp.get(); + completed = completedResp.get(); + failed = failedResp.get(); Map out = new LinkedHashMap<>(); out.put("enqueued_total", enqueuedTotal); out.put("completed_total", completedTotal); @@ -308,7 +299,7 @@ Before running the demo, make sure that: * Redis 6.2 or later running locally on the default port (6379). `BLMOVE` requires Redis 6.2+; on older servers, replace the call with `BRPOPLPUSH`. * Java 17 or later (the demo uses text-block-free string concatenation but still relies on a modern JDK). -* Jedis 5.x is on the classpath. The smallest workable classpath is the Jedis jar plus its two transitive dependencies, `commons-pool2` and `slf4j-api`. +* Jedis 8.x is on the classpath, along with its runtime dependencies: `commons-pool2`, `slf4j-api`, `gson`, `json`, and `redis-authx-core`. If you use Maven: @@ -316,14 +307,14 @@ If you use Maven: redis.clients jedis - 5.0.1 + 8.0.2 ``` If you use Gradle: ```groovy -implementation 'redis.clients:jedis:5.0.1' +implementation 'redis.clients:jedis:8.0.2' ``` ## Running the demo @@ -347,10 +338,10 @@ curl -O $BASE/JsonUtil.java A local demo server is included to show the queue in action ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/job-queue/java-jedis/DemoServer.java)). Compile and run with `javac` and `java`, listing each jar on the classpath: ```bash -javac -cp jedis-5.0.1.jar:commons-pool2-2.12.1.jar:slf4j-api-2.0.12.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 RedisJobQueue.java JobWorker.java WorkerPool.java DemoServer.java -java -cp .:jedis-5.0.1.jar:commons-pool2-2.12.1.jar:slf4j-api-2.0.12.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 8793 --visibility-ms 5000 ``` @@ -391,9 +382,9 @@ The visibility timeout has to exceed the longest real job time, with margin. If The demo only reclaims when you click the button. In production, run `reclaimStuck()` from a `ScheduledExecutorService` (every few seconds for fast queues, every minute for slow ones), or from each worker before it blocks on `claim()`. Both patterns work as long as *someone* runs the sweep. -### Size `JedisPool` for your worker count +### Size the connection pool for your worker count -`JedisPool` is thread-safe and connections are released back to the pool when the try-with-resources block exits. The demo bumps `maxTotal` to 32 to support the blocking `BLMOVE` call held by each worker plus the per-request connections used by the HTTP handlers. As a rule of thumb, `maxTotal` should be at least *workers + concurrent HTTP request threads + reclaimer threads + headroom*. +`RedisClient` is thread-safe and connections are released back to the pool when each call finishes. The demo builds the client with a `ConnectionPoolConfig` and bumps `maxTotal` to 32 to support the blocking `BLMOVE` call held by each worker plus the per-request connections used by the HTTP handlers. As a rule of thumb, `maxTotal` should be at least *workers + concurrent HTTP request threads + reclaimer threads + headroom*. ### Use a separate Redis database or key prefix per queue