From 35e2865213f7546eab7c1d9eaaae08ef9457c95d Mon Sep 17 00:00:00 2001 From: seakider Date: Sat, 28 Dec 2024 15:28:18 +0800 Subject: [PATCH] Fixed - TasksService.addAsync() and RedissonExecutorService.pushTaskAsync(), 'publish' won't take effect #6365 Signed-off-by: xuxiaolei --- .../src/main/java/org/redisson/RedissonExecutorService.java | 2 +- .../src/main/java/org/redisson/executor/TasksService.java | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/redisson/src/main/java/org/redisson/RedissonExecutorService.java b/redisson/src/main/java/org/redisson/RedissonExecutorService.java index d107f4bbe..d50e16638 100644 --- a/redisson/src/main/java/org/redisson/RedissonExecutorService.java +++ b/redisson/src/main/java/org/redisson/RedissonExecutorService.java @@ -260,7 +260,7 @@ public class RedissonExecutorService implements RScheduledExecutorService { + "local v = redis.call('zrange', KEYS[2], 0, 0); " // if new task added to queue head then publish its startTime // to all scheduler workers - + "if v[1] == expiredTaskIds[i] then " + + "if v[1] == scheduledName then " + "redis.call('publish', KEYS[3], startTime); " + "end;" diff --git a/redisson/src/main/java/org/redisson/executor/TasksService.java b/redisson/src/main/java/org/redisson/executor/TasksService.java index 796c0c403..70b7e5e4b 100644 --- a/redisson/src/main/java/org/redisson/executor/TasksService.java +++ b/redisson/src/main/java/org/redisson/executor/TasksService.java @@ -132,12 +132,13 @@ public class TasksService extends BaseRemoteService { + "end; " + "if tonumber(ARGV[1]) > 0 then " + + "local scheduledName = 'ff:' .. ARGV[2];" + "redis.call('set', KEYS[7], ARGV[4]);" - + "redis.call('zadd', KEYS[3], ARGV[1], 'ff:' .. ARGV[2]);" + + "redis.call('zadd', KEYS[3], ARGV[1], scheduledName);" + "local v = redis.call('zrange', KEYS[3], 0, 0); " // if new task added to queue head then publish its startTime // to all scheduler workers - + "if v[1] == ARGV[2] then " + + "if v[1] == scheduledName then " + "redis.call('publish', KEYS[4], ARGV[1]); " + "end; " + "end;"