diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/BatchJobConsumerDemo.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/AdvancedJobConsumerBatchPopDemo.java similarity index 87% rename from examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/BatchJobConsumerDemo.java rename to examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/AdvancedJobConsumerBatchPopDemo.java index 14bfa46b..3b40f5c4 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/BatchJobConsumerDemo.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/AdvancedJobConsumerBatchPopDemo.java @@ -7,7 +7,7 @@ import java.util.concurrent.TimeUnit; import org.springside.examples.showcase.demos.redis.JedisPoolFactory; import org.springside.modules.nosql.redis.JedisUtils; -import org.springside.modules.nosql.redis.scheduler.BatchJobConsumer; +import org.springside.modules.nosql.redis.scheduler.AdvancedConsumer; import org.springside.modules.nosql.redis.scheduler.SimpleJobConsumer; import org.springside.modules.test.benchmark.ConcurrentBenchmark; import org.springside.modules.utils.Threads; @@ -19,9 +19,9 @@ import org.springside.modules.utils.Threads; * * @author calvin */ -public class BatchJobConsumerDemo extends SimpleJobConsumerDemo { +public class AdvancedJobConsumerBatchPopDemo extends SimpleJobConsumerDemo { - private BatchJobConsumer consumer; + private AdvancedConsumer consumer; private static boolean reliable; private static int batchSize; @@ -32,9 +32,9 @@ public class BatchJobConsumerDemo extends SimpleJobConsumerDemo { String.valueOf(THREAD_COUNT))); reliable = Boolean.parseBoolean(System.getProperty("reliable", - String.valueOf(BatchJobConsumer.DEFAULT_RELIABLE))); + String.valueOf(AdvancedConsumer.DEFAULT_RELIABLE))); batchSize = Integer.parseInt(System.getProperty("batchsize", - String.valueOf(BatchJobConsumer.DEFAULT_BATCH_SIZE))); + String.valueOf(AdvancedConsumer.DEFAULT_BATCH_SIZE))); pool = JedisPoolFactory.createJedisPool(JedisUtils.DEFAULT_HOST, JedisUtils.DEFAULT_PORT, JedisUtils.DEFAULT_TIMEOUT, threadCount); @@ -68,8 +68,8 @@ public class BatchJobConsumerDemo extends SimpleJobConsumerDemo { } } - public BatchJobConsumerDemo() { - consumer = new BatchJobConsumer("ss", pool); + public AdvancedJobConsumerBatchPopDemo() { + consumer = new AdvancedConsumer("ss", pool); consumer.setReliable(reliable); consumer.setBatchSize(batchSize); } diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/ReliableJobConsumerDemo.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/AdvancedJobConsumerSinglePopDemo.java similarity index 87% rename from examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/ReliableJobConsumerDemo.java rename to examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/AdvancedJobConsumerSinglePopDemo.java index 658b3fe0..5b3533e2 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/ReliableJobConsumerDemo.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/AdvancedJobConsumerSinglePopDemo.java @@ -6,7 +6,7 @@ import java.util.concurrent.TimeUnit; import org.springside.examples.showcase.demos.redis.JedisPoolFactory; import org.springside.modules.nosql.redis.JedisUtils; -import org.springside.modules.nosql.redis.scheduler.ReliableJobConsumer; +import org.springside.modules.nosql.redis.scheduler.AdvancedConsumer; import org.springside.modules.nosql.redis.scheduler.SimpleJobConsumer; import org.springside.modules.test.benchmark.ConcurrentBenchmark; import org.springside.modules.utils.Threads; @@ -18,9 +18,9 @@ import org.springside.modules.utils.Threads; * * @author calvin */ -public class ReliableJobConsumerDemo extends SimpleJobConsumerDemo { +public class AdvancedJobConsumerSinglePopDemo extends SimpleJobConsumerDemo { - private ReliableJobConsumer consumer; + private AdvancedConsumer consumer; public static void main(String[] args) throws Exception { @@ -59,8 +59,9 @@ public class ReliableJobConsumerDemo extends SimpleJobConsumerDemo { } } - public ReliableJobConsumerDemo() { - consumer = new ReliableJobConsumer("ss", pool); + public AdvancedJobConsumerSinglePopDemo() { + consumer = new AdvancedConsumer("ss", pool); + consumer.setReliable(true); } @Override diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/AdvancedConsumer.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/AdvancedConsumer.java new file mode 100644 index 00000000..844bf8c1 --- /dev/null +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/AdvancedConsumer.java @@ -0,0 +1,145 @@ +/*------------------------------------------------------------------------------ + * COPYRIGHT Ericsson 2013 + * + * The copyright to the computer program(s) herein is the property of + * Ericsson Inc. The programs may be used and/or copied only with written + * permission from Ericsson Inc. or in accordance with the terms and + * conditions stipulated in the agreement/contract under which the + * program(s) have been supplied. + *----------------------------------------------------------------------------*/ +package org.springside.modules.nosql.redis.scheduler; + +import java.util.List; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springside.modules.nosql.redis.JedisScriptExecutor; +import org.springside.modules.nosql.redis.JedisTemplate; +import org.springside.modules.utils.Threads; + +import redis.clients.jedis.JedisPool; +import redis.clients.jedis.exceptions.JedisConnectionException; + +import com.google.common.collect.Lists; + +/** + * 高级的使用Lua脚本取回任务，支持高可靠性和批量取回任务，但不会阻塞，如果没有任务即时返回。 + * + * 在高可靠模式下，任务在返回给客户端的同时，会放入lock table中，客户完成任务后必须调用ack()删除任务，否则Dispatcher会将超时未完成的任务放入队列重新执行. + * + * @author calvin + */ +public class AdvancedConsumer { + public static final String DEFAULT_BATCH_POP_LUA_FILE_PATH = "classpath:/redis/batchpop.lua"; + public static final String DEFAULT_SINGLE_POP_LUA_FILE_PATH = "classpath:/redis/singlepop.lua"; + + public static final int DEFAULT_CONNECTION_RETRY_MILLS = 5000; + public static final boolean DEFAULT_RELIABLE = false; + public static final int DEFAULT_BATCH_SIZE = 10; + + private static Logger logger = LoggerFactory.getLogger(AdvancedConsumer.class); + + private boolean reliable = DEFAULT_RELIABLE; + private int batchSize = DEFAULT_BATCH_SIZE; + + private JedisTemplate jedisTemplate; + private JedisScriptExecutor singlePopScriptExecutor; + private JedisScriptExecutor batchPopScriptExecutor; + + private String batchPopScriptPath = DEFAULT_BATCH_POP_LUA_FILE_PATH; + private String singlePopScriptPath = DEFAULT_SINGLE_POP_LUA_FILE_PATH; + + private String readyJobKey; + private String lockJobKey; + private List keys; + + public AdvancedConsumer(String jobName, JedisPool jedisPool) { + readyJobKey = Keys.getReadyJobKey(jobName); + lockJobKey = Keys.getLockJobKey(jobName); + keys = Lists.newArrayList(readyJobKey, lockJobKey); + + jedisTemplate = new JedisTemplate(jedisPool); + singlePopScriptExecutor = new JedisScriptExecutor(jedisPool); + batchPopScriptExecutor = new JedisScriptExecutor(jedisPool); + } + + /** + * 初始化脚本，在popup前必须被调用. + */ + public void init() { + singlePopScriptExecutor.loadFromFile(singlePopScriptPath); + batchPopScriptExecutor.loadFromFile(batchPopScriptPath); + } + + /** + * 即时返回任务, 如有任务返回的同时将其放入lock job set，如无任务返回null. + * 如发生redis连接异常, 线程会sleep 5秒后返回null， + * 如果发生redis数据错误如lua脚本错误，抛出异常. + */ + public String popupJob() { + String job = null; + try { + long currTime = System.currentTimeMillis(); + List args = Lists.newArrayList(String.valueOf(currTime), String.valueOf(reliable)); + job = (String) singlePopScriptExecutor.execute(keys, args); + } catch (JedisConnectionException e) { + Threads.sleep(DEFAULT_CONNECTION_RETRY_MILLS); + } + + return job; + } + + /** + * 即时批量跑回任务, 如有任务返回的同时将其放入lock job set，如无任务返回空的List. + * 如发生redis连接异常, 线程会sleep 5秒后返回null， + * 如果发生redis数据错误如lua脚本错误，抛出异常. + */ + public List popupJobs() { + List jobs = null; + try { + long currTime = System.currentTimeMillis(); + List args = Lists.newArrayList(String.valueOf(currTime), String.valueOf(batchSize), + String.valueOf(reliable)); + jobs = (List) batchPopScriptExecutor.execute(keys, args); + } catch (JedisConnectionException e) { + Threads.sleep(DEFAULT_CONNECTION_RETRY_MILLS); + } + + return jobs; + } + + /** + * 在高可靠模式下，报告任务完成, 从lock table set中删除任务. + */ + public void ackJob(String job) { + jedisTemplate.zrem(lockJobKey, job); + } + + /** + * 设置不在默认路径的lua script path，按Spring Resource的URL格式. + */ + public void setBatchPopScriptPath(String batchPopScriptPath) { + this.batchPopScriptPath = batchPopScriptPath; + } + + /** + * 设置不在默认路径的lua script path，按Spring Resource的URL格式. + */ + public void setSinglePopScriptPath(String singlePopScriptPath) { + this.singlePopScriptPath = singlePopScriptPath; + } + + /** + * 设置是否高可靠模式。 + */ + public void setReliable(boolean reliable) { + this.reliable = reliable; + } + + /** + * 设置批量取回任务数量. + */ + public void setBatchSize(int batchSize) { + this.batchSize = batchSize; + } +} diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/BatchJobConsumer.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/BatchJobConsumer.java deleted file mode 100644 index b57049ce..00000000 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/BatchJobConsumer.java +++ /dev/null @@ -1,99 +0,0 @@ -/*------------------------------------------------------------------------------ - * COPYRIGHT Ericsson 2013 - * - * The copyright to the computer program(s) herein is the property of - * Ericsson Inc. The programs may be used and/or copied only with written - * permission from Ericsson Inc. or in accordance with the terms and - * conditions stipulated in the agreement/contract under which the - * program(s) have been supplied. - *----------------------------------------------------------------------------*/ -package org.springside.modules.nosql.redis.scheduler; - -import java.util.List; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springside.modules.nosql.redis.JedisScriptExecutor; -import org.springside.modules.nosql.redis.JedisTemplate; -import org.springside.modules.utils.Threads; - -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.exceptions.JedisConnectionException; - -import com.google.common.collect.Lists; - -public class BatchJobConsumer { - public static final int DEFAULT_CONNECTION_RETRY_MILLS = 5000; - public static final String DEFAULT_LUA_FILE_PATH = "classpath:/redis/batchpop.lua"; - - public static final boolean DEFAULT_RELIABLE = false; - public static final int DEFAULT_BATCH_SIZE = 1; - - private static Logger logger = LoggerFactory.getLogger(BatchJobConsumer.class); - - private boolean reliable = DEFAULT_RELIABLE; - private int batchSize = DEFAULT_BATCH_SIZE; - - private JedisTemplate jedisTemplate; - private JedisScriptExecutor scriptExecutor; - - private String scriptPath = DEFAULT_LUA_FILE_PATH; - private String readyJobKey; - private String lockJobKey; - private List keys; - - public BatchJobConsumer(String jobName, JedisPool jedisPool) { - readyJobKey = Keys.getReadyJobKey(jobName); - lockJobKey = Keys.getLockJobKey(jobName); - keys = Lists.newArrayList(readyJobKey, lockJobKey); - - jedisTemplate = new JedisTemplate(jedisPool); - scriptExecutor = new JedisScriptExecutor(jedisPool); - } - - public void init() { - scriptExecutor.loadFromFile(scriptPath); - } - - public List popupJobs() { - List jobs = null; - try { - long currTime = System.currentTimeMillis(); - List args = Lists.newArrayList(String.valueOf(currTime), String.valueOf(batchSize), - String.valueOf(reliable)); - jobs = (List) scriptExecutor.execute(keys, args); - } catch (JedisConnectionException e) { - Threads.sleep(DEFAULT_CONNECTION_RETRY_MILLS); - } - - return jobs; - } - - /** - * 在高可靠模式下，报告任务完成, 从lock table set中删除. - */ - public void ackJob(String job) { - jedisTemplate.zrem(lockJobKey, job); - } - - /** - * 设置不在默认路径的lua script path，按Spring Resource的URL格式. - */ - public void setScriptPath(String scriptPath) { - this.scriptPath = scriptPath; - } - - /** - * 设置是否高可靠模式。 - */ - public void setReliable(boolean reliable) { - this.reliable = reliable; - } - - /** - * 设置批量取回任务数量. - */ - public void setBatchSize(int batchSize) { - this.batchSize = batchSize; - } -} diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobProducer.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobProducer.java index 2b2c6831..5b40579a 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobProducer.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobProducer.java @@ -44,18 +44,6 @@ public class JobProducer { jedisTemplate.lpush(readyJobKey, job); } - /** - * 尝试取消立即执行的任务,如果任务不存在或已被执行返回false, 否则返回true. - */ - public boolean cancel(final String job) { - boolean removed = jedisTemplate.lremOne(readyJobKey, job); - if (!removed) { - logger.warn("Can't cancel job by value {}", job); - } - - return removed; - } - /** * 安排延时执行任务. */ @@ -67,7 +55,7 @@ public class JobProducer { /** * 尝试取消延时任务, 如果任务不存在或已被触发返回false, 否则返回true. */ - public boolean cancelSchedule(final String job) { + public boolean cancel(final String job) { boolean removed = jedisTemplate.zrem(scheduledJobKey, job); if (!removed) { diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobStatistics.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobStatistics.java index 014348f0..3537eb8f 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobStatistics.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobStatistics.java @@ -31,42 +31,42 @@ public class JobStatistics { } /** - * 获取未达到触发条件进行分发的Job数量. + * 获取已安排但未达到触发条件的Job数量. */ public long getScheduledJobNumber() { return jedisTemplate.zcard(scheduledJobKey); } /** - * 获取已分发但未被执行的Job数量. + * 获取已触发但未被客户端取走的Job数量. */ public long getReadyJobNumber() { return jedisTemplate.llen(readyJobKey); } /** - * 获取高可靠模式下已被取走执行但未报告完成的Job数量. + * 获取高可靠模式下，已被取走执行但未报告完成的Job数量. */ public long getLockJobNumber() { return jedisTemplate.zcard(lockJobKey); } /** - * 获取已分发的Job总数。 + * 获取已触发的Job总数。 */ public long getDispatchCounter() { return jedisTemplate.getAsLong(dispatchCounterKey); } /** - * 获取已重做的Job总数。 + * 获取高可靠模式下，已重做的Job总数。 */ public long getRetryCounter() { return jedisTemplate.getAsLong(retryCounterKey); } /** - * 重置已分发的Job计数器和已重做的计数器. + * 重置所有计数器. */ public void restCounters() { jedisTemplate.set(dispatchCounterKey, "0"); diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/ReliableJobConsumer.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/ReliableJobConsumer.java deleted file mode 100644 index 5bc7b254..00000000 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/ReliableJobConsumer.java +++ /dev/null @@ -1,84 +0,0 @@ -/*------------------------------------------------------------------------------ - * COPYRIGHT Ericsson 2013 - * - * The copyright to the computer program(s) herein is the property of - * Ericsson Inc. The programs may be used and/or copied only with written - * permission from Ericsson Inc. or in accordance with the terms and - * conditions stipulated in the agreement/contract under which the - * program(s) have been supplied. - *----------------------------------------------------------------------------*/ -package org.springside.modules.nosql.redis.scheduler; - -import java.util.List; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springside.modules.nosql.redis.JedisScriptExecutor; -import org.springside.modules.nosql.redis.JedisTemplate; -import org.springside.modules.utils.Threads; - -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.exceptions.JedisConnectionException; - -import com.google.common.collect.Lists; - -public class ReliableJobConsumer { - - public static final int DEFAULT_CONNECTION_RETRY_MILLS = 5000; - public static final String DEFAULT_LUA_FILE_PATH = "classpath:/redis/reliablepop.lua"; - - private static Logger logger = LoggerFactory.getLogger(ReliableJobConsumer.class); - - private JedisTemplate jedisTemplate; - private JedisScriptExecutor scriptExecutor; - - private String scriptPath = DEFAULT_LUA_FILE_PATH; - private String readyJobKey; - private String lockJobKey; - private List keys; - - public ReliableJobConsumer(String jobName, JedisPool jedisPool) { - readyJobKey = Keys.getReadyJobKey(jobName); - lockJobKey = Keys.getLockJobKey(jobName); - keys = Lists.newArrayList(readyJobKey, lockJobKey); - - jedisTemplate = new JedisTemplate(jedisPool); - scriptExecutor = new JedisScriptExecutor(jedisPool); - } - - public void init() { - scriptExecutor.loadFromFile(scriptPath); - } - - /** - * 即时返回任务, 如有任务返回的同时将其放入lock job set，如无任务返回null. - * 如发生redis连接异常, 线程会sleep 5秒后返回null， - * 如果发生redis数据错误如lua脚本错误，抛出异常. - */ - public String popupJob() { - String job = null; - try { - long currTime = System.currentTimeMillis(); - List args = Lists.newArrayList(String.valueOf(currTime)); - job = (String) scriptExecutor.execute(keys, args); - } catch (JedisConnectionException e) { - Threads.sleep(DEFAULT_CONNECTION_RETRY_MILLS); - } - - return job; - } - - /** - * 报告任务完成, 从lock table set中删除. - */ - public void ackJob(String job) { - jedisTemplate.zrem(lockJobKey, job); - } - - /** - * 设置不在默认路径的lua script path，按Spring Resource的URL格式. - */ - public void setScriptPath(String scriptPath) { - this.scriptPath = scriptPath; - } -} diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/SimpleJobConsumer.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/SimpleJobConsumer.java index a91b535c..6a0c490b 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/SimpleJobConsumer.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/SimpleJobConsumer.java @@ -22,17 +22,19 @@ import redis.clients.jedis.JedisPool; import redis.clients.jedis.exceptions.JedisConnectionException; /* - * 简单的阻塞取出任务。 + * 简单的基于brpop()API, 阻塞的取出任务。 + * brpop的阻塞，在线程中断时不会自动退出，所以还是设置有限timeout时间，另外在线程池退出时已比timeout时间长的时间调用awaitTermination()等待线程结束. */ public class SimpleJobConsumer { - public static final int DEFAULT_CONNECTION_RETRY_MILLS = 5000; public static final int DEFAULT_POPUP_TIMEOUT_SECONDS = 5; + public static final int DEFAULT_CONNECTION_RETRY_MILLS = 5000; private static Logger logger = LoggerFactory.getLogger(SimpleJobConsumer.class); private JedisTemplate jedisTemplate; private String readyJobKey; + private int popupTimeoutSecs = DEFAULT_POPUP_TIMEOUT_SECONDS; public SimpleJobConsumer(String jobName, JedisPool jedisPool) { jedisTemplate = new JedisTemplate(jedisPool); @@ -40,7 +42,7 @@ public class SimpleJobConsumer { } /** - * 阻塞直到返回任务，如果五5秒内无任务到达，返回null. + * 阻塞直到返回任务，如果popupTimeoutSecs内(默认5秒)无任务到达，返回null. * 如发生redis连接异常, 线程会sleep 5秒后返回null， */ public String popupJob() { @@ -50,7 +52,7 @@ public class SimpleJobConsumer { nameValuePair = jedisTemplate.execute(new JedisAction>() { @Override public List action(Jedis jedis) { - return jedis.brpop(DEFAULT_POPUP_TIMEOUT_SECONDS, readyJobKey); + return jedis.brpop(popupTimeoutSecs, readyJobKey); } }); } catch (JedisConnectionException e) { @@ -63,4 +65,8 @@ public class SimpleJobConsumer { return null; } } + + public void setPopupTimeoutSecs(int popupTimeoutSecs) { + this.popupTimeoutSecs = popupTimeoutSecs; + } } diff --git a/modules/extension/src/main/resources/redis/batchpop.lua b/modules/extension/src/main/resources/redis/batchpop.lua index a7114a79..8e6858ae 100644 --- a/modules/extension/src/main/resources/redis/batchpop.lua +++ b/modules/extension/src/main/resources/redis/batchpop.lua @@ -3,10 +3,10 @@ local batchJobs={} for i=1,ARGV[2] do local job = redis.call('rpop', KEYS[1]) - if (job == nil) then - break - else + if (job) then batchJobs[i] = job + else + break end end diff --git a/modules/extension/src/main/resources/redis/dispatch.lua b/modules/extension/src/main/resources/redis/dispatch.lua index 2145ad8f..b4bad36d 100644 --- a/modules/extension/src/main/resources/redis/dispatch.lua +++ b/modules/extension/src/main/resources/redis/dispatch.lua @@ -3,47 +3,50 @@ local batchsize = 64 -local function addReadyJob(jobs, count) - local i = 1 - while i<=count do - local j = 1 - local batchJobs={} - while i<=count and j<=batchsize do - batchJobs[j]= jobs[i] - i=i+1 - j=j+1 - end - redis.call('lpush', KEYS[2], unpack(batchJobs)) - end +local function batchAddReadyJob(jobs, count) + local i = 1 + while i<=count do + local j = 1 + local batchJobs={} + while i<=count and j<=batchsize do + batchJobs[j]= jobs[i] + i=i+1 + j=j+1 + end + redis.call('lpush', KEYS[2], unpack(batchJobs)) + end end -local scheduledjobs=redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1]) +-- Comments: find triggered scheduled jobs +local scheduledjobs = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1]) local scheduledJobCount = table.maxn(scheduledjobs) if scheduledJobCount>0 then - -- Comments: add to the Ready Job list - addReadyJob(scheduledjobs, scheduledJobCount) + -- Comments: add to the Ready Job list + batchAddReadyJob(scheduledjobs, scheduledJobCount) - -- Comments: remove from scheduled Job sorted set - redis.call('zremrangebyscore', KEYS[1], '-inf', ARGV[1]) - - -- Comments: incr Dispatch counter - redis.call('incrby', KEYS[3], scheduledJobCount) + -- Comments: remove from Scheduled Job sorted set + redis.call('zremrangebyscore', KEYS[1], '-inf', ARGV[1]) + + -- Comments: incr Dispatch counter + redis.call('incrby', KEYS[3], scheduledJobCount) end +-- Comments: reliable mode if (ARGV[2] == 'true') then + -- Comments: find timeout job from Lock Job sorted set local expiredTime = tonumber(ARGV[1]) - tonumber(ARGV[3]) local lockJobs = redis.call('zrangebyscore', KEYS[4], '-inf', expiredTime) local lockJobCount = table.maxn(lockJobs) - - if lockJobCount>0 then - -- Comments: add to the Ready Job list - addReadyJob(lockJobs, lockJobCount) - -- Comments: remove from scheduled Job sorted set - redis.call('zremrangebyscore', KEYS[4], '-inf', expiredTime) - - -- Comments: incr Retry Lock counter - redis.call('incrby', KEYS[5], lockJobCount) + if lockJobCount > 0 then + -- Comments: add to the Ready Job list + addReadyJob(lockJobs, lockJobCount) + + -- Comments: remove from Lock Job sorted set + redis.call('zremrangebyscore', KEYS[4], '-inf', expiredTime) + + -- Comments: incr Retry Lock counter + redis.call('incrby', KEYS[5], lockJobCount) end end diff --git a/modules/extension/src/main/resources/redis/reliablepop.lua b/modules/extension/src/main/resources/redis/singlepop.lua similarity index 64% rename from modules/extension/src/main/resources/redis/reliablepop.lua rename to modules/extension/src/main/resources/redis/singlepop.lua index 3b3ab123..6122f766 100644 --- a/modules/extension/src/main/resources/redis/reliablepop.lua +++ b/modules/extension/src/main/resources/redis/singlepop.lua @@ -1,9 +1,9 @@ -- KEYS: [1]job:ready, [2]job:lock --- ARGV: [1]currentTime +-- ARGV: [1]currentTime, [2]reliable local job=redis.call('rpop', KEYS[1]) -if (job ~=nil) then +if (job and ARGV[2] == 'true') then redis.call('zadd', KEYS[2], ARGV[1], job) end