From 7cc72a0190d07560b0eda159d959ff26acd80b8a Mon Sep 17 00:00:00 2001 From: Calvin Date: Fri, 12 Jul 2013 01:50:38 +0800 Subject: [PATCH] =?UTF-8?q?#221=20=E5=8A=A0=E5=85=A5=E5=AF=B9Jedis?= =?UTF-8?q?=E7=9A=84=E5=B0=81=E8=A3=85=EF=BC=8C=E4=BB=A3=E7=A0=81=E6=94=B9?= =?UTF-8?q?=E8=BF=9B=E4=B8=8E=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../demos/redis/job/JobListenerDemo.java | 2 +- .../nosql/redis/JedisScriptExecutor.java | 13 ++++-- .../nosql/redis/scheduler/JobDispatcher.java | 41 ++++++++++++------- .../nosql/redis/scheduler/JobListener.java | 12 +++--- .../nosql/redis/scheduler/JobManager.java | 10 ++--- .../modules/nosql/redis/scheduler/Keys.java | 12 ++++++ 6 files changed, 61 insertions(+), 29 deletions(-) create mode 100644 modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/Keys.java diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobListenerDemo.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobListenerDemo.java index 8fb42d5c..733f5159 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobListenerDemo.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobListenerDemo.java @@ -53,7 +53,7 @@ public class JobListenerDemo implements JobHandler { if (c == '\n') { System.out.println("Shutting down"); threadPool.shutdownNow(); - boolean shutdownSucess = threadPool.awaitTermination(JobListener.POPUP_TIMEOUT + 1, + boolean shutdownSucess = threadPool.awaitTermination(JobListener.DEFAULT_POPUP_TIMEOUT + 1, TimeUnit.SECONDS); if (!shutdownSucess) { diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/JedisScriptExecutor.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/JedisScriptExecutor.java index ff6d3a8a..fd8b6415 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/JedisScriptExecutor.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/JedisScriptExecutor.java @@ -58,6 +58,9 @@ public class JedisScriptExecutor { */ public Object execute(final String hash, final List keys, final List args) throws IllegalArgumentException { + + final long start = System.currentTimeMillis(); + if (!hashScriptMap.containsKey(hash)) { throw new IllegalArgumentException("Script hash " + hash + " is not loaded in executor。"); } @@ -66,12 +69,16 @@ public class JedisScriptExecutor { return jedisTemplate.execute(new JedisAction() { @Override public Object action(Jedis jedis) { - return jedis.evalsha(hash, keys, args); + Object result = jedis.evalsha(hash, keys, args); + logger.debug("Script hash {} execution time is {}ms", hash, System.currentTimeMillis() - start); + return result; } }); } catch (JedisDataException e) { - logger.warn("Lua execution error, try to reload the script.", e); - return reloadAndExecute(hash, keys, args); + logger.warn("Script hash {} is not loaded yet, try to reload and run it again", hash, e); + Object result = reloadAndExecute(hash, keys, args); + logger.debug("Script hash {} reload and execution time is {}ms", hash, System.currentTimeMillis() - start); + return result; } } diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobDispatcher.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobDispatcher.java index 01ca06b7..b0dce905 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobDispatcher.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobDispatcher.java @@ -11,7 +11,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.core.io.DefaultResourceLoader; import org.springframework.core.io.Resource; -import org.springframework.core.io.ResourceLoader; import org.springside.modules.nosql.redis.JedisScriptExecutor; import org.springside.modules.utils.Threads.WrapExceptionRunnable; @@ -19,12 +18,17 @@ import redis.clients.jedis.JedisPool; import com.google.common.collect.Lists; +/** + * 定时分发任务。 启动线程定时从sleeping job sorted set 中取出到期的任务放入ready job list. + * + * @author calvin + */ public class JobDispatcher implements Runnable { public static final String DEFAULT_DISPATCH_LUA_FILE = "classpath:/redis/dispatch.lua"; private static Logger logger = LoggerFactory.getLogger(JobDispatcher.class); - private ScheduledExecutorService threadPool; + private ScheduledExecutorService scheduledThreadPool; private JedisScriptExecutor scriptExecutor; @@ -37,31 +41,39 @@ public class JobDispatcher implements Runnable { } public JobDispatcher(String jobName, JedisPool jedisPool, String scriptPath) { + keys = Lists.newArrayList(Keys.getSleepingJobKey(jobName), Keys.getReadyJobKey(jobName)); this.scriptExecutor = new JedisScriptExecutor(jedisPool); - keys = Lists.newArrayList(jobName + ".job:sleeping", jobName + ".job:ready"); loadLuaScript(scriptPath); } private void loadLuaScript(String scriptPath) { - ResourceLoader resourceLoader = new DefaultResourceLoader(); - Resource resource = resourceLoader.getResource(scriptPath); + String script; try { - String script = FileUtils.readFileToString(resource.getFile()); - scriptHash = scriptExecutor.load(script); + Resource resource = new DefaultResourceLoader().getResource(scriptPath); + script = FileUtils.readFileToString(resource.getFile()); } catch (IOException e) { - throw new IllegalStateException(DEFAULT_DISPATCH_LUA_FILE + "not exist", e); + throw new IllegalArgumentException(scriptPath + " is not exist.", e); } + + scriptHash = scriptExecutor.load(script); } + /** + * 启动分发线程, 自行创建scheduler线程池. + */ public void start(long periodMilliseconds) { - threadPool = Executors.newScheduledThreadPool(1); - threadPool.scheduleAtFixedRate(new WrapExceptionRunnable(this), 0, periodMilliseconds, TimeUnit.MILLISECONDS); + this.scheduledThreadPool = Executors.newScheduledThreadPool(1); + scheduledThreadPool.scheduleAtFixedRate(new WrapExceptionRunnable(this), 0, periodMilliseconds, + TimeUnit.MILLISECONDS); } + /** + * 停止分发任务, 默认最多延时10秒等候线程关闭. + */ public void stop() { - threadPool.shutdownNow(); + scheduledThreadPool.shutdownNow(); try { - if (!threadPool.awaitTermination(5, TimeUnit.SECONDS)) { + if (!scheduledThreadPool.awaitTermination(10, TimeUnit.SECONDS)) { logger.error("Job dispatcher terminate failed!"); } } catch (InterruptedException e) { @@ -69,12 +81,13 @@ public class JobDispatcher implements Runnable { } } + /** + * 以当前时间为参数执行Lua Script分发任务。 + */ @Override public void run() { long currTime = System.currentTimeMillis(); List args = Lists.newArrayList(String.valueOf(currTime)); scriptExecutor.execute(scriptHash, keys, args); - long luaExecTime = System.currentTimeMillis() - currTime; - logger.debug("Execution Time={}ms.", luaExecTime); } } diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobListener.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobListener.java index b2df331a..48cd28ea 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobListener.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobListener.java @@ -18,21 +18,21 @@ import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; /** - * This is the Redis implementation of SchedulerManager. + * 阻塞接收任务的Runnable. */ public class JobListener implements Runnable { - public static final int POPUP_TIMEOUT = 5; + public static final int DEFAULT_POPUP_TIMEOUT = 5; - private String readyJobName; + private String readyJobKey; - private JedisTemplate jedisTemplate = null; + private JedisTemplate jedisTemplate; private final JobHandler jobHandler; public JobListener(String jobName, JedisPool jedisPool, JobHandler jobHandler) { jedisTemplate = new JedisTemplate(jedisPool); - readyJobName = jobName + ".job:ready"; + readyJobKey = Keys.getReadyJobKey(jobName); this.jobHandler = jobHandler; } @@ -42,7 +42,7 @@ public class JobListener implements Runnable { @Override public void action(Jedis jedis) { while (!Thread.currentThread().isInterrupted()) { - List nameValuePair = jedis.brpop(POPUP_TIMEOUT, readyJobName); + List nameValuePair = jedis.brpop(DEFAULT_POPUP_TIMEOUT, readyJobKey); if ((nameValuePair != null) && !nameValuePair.isEmpty()) { String job = nameValuePair.get(1); jobHandler.handleJob(job); diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobManager.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobManager.java index 032f7688..e2d9b348 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobManager.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobManager.java @@ -24,13 +24,13 @@ public class JobManager { private static Logger logger = LoggerFactory.getLogger(JobManager.class); - private String sleepingJobName; + private String sleepingJobKey; private JedisTemplate jedisTemplate; public JobManager(String jobName, JedisPool jedisPool) { jedisTemplate = new JedisTemplate(jedisPool); - sleepingJobName = jobName + ".job:sleeping"; + sleepingJobKey = Keys.getSleepingJobKey(jobName); } /** @@ -38,14 +38,14 @@ public class JobManager { */ public void scheduleJob(final String job, final long delay, final TimeUnit timeUnit) { final long delayTimeInMillisecond = System.currentTimeMillis() + timeUnit.toMillis(delay); - jedisTemplate.zadd(sleepingJobName, delayTimeInMillisecond, job); + jedisTemplate.zadd(sleepingJobKey, delayTimeInMillisecond, job); } /** - * 取消任务,如果任务不存在或已触发返回false, 否则返回true. + * 取消任务, 如果任务不存在或已被触发返回false, 否则返回true. */ public boolean cancelJob(final String job) { - boolean removed = jedisTemplate.zrem(sleepingJobName, job); + boolean removed = jedisTemplate.zrem(sleepingJobKey, job); if (!removed) { logger.warn("Can't cancel job by value {}", job); diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/Keys.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/Keys.java new file mode 100644 index 00000000..d24756c1 --- /dev/null +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/Keys.java @@ -0,0 +1,12 @@ +package org.springside.modules.nosql.redis.scheduler; + +public class Keys { + + public static String getSleepingJobKey(String jobName) { + return new StringBuilder().append("job:").append(jobName).append(":sleepingjob").toString(); + } + + public static String getReadyJobKey(String jobName) { + return new StringBuilder().append("job:").append(jobName).append(":sleepingjob").toString(); + } +}