From 7bbf836fbf6cf980c823800f944be5597fcbdf1a Mon Sep 17 00:00:00 2001 From: Calvin Date: Sun, 14 Jul 2013 23:35:49 +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=E6=94=B9=E5=8F=98=E5=87=BD?= =?UTF-8?q?=E6=95=B0=E5=AE=9A=E4=B9=89=E6=9B=B4=E6=96=B9=E4=BE=BFSpring?= =?UTF-8?q?=E6=B3=A8=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../redis/elector/MasterElectorDemo.java | 4 +- .../demos/redis/job/JobDispatcherDemo.java | 2 +- .../nosql/redis/elector/MasterElector.java | 29 +++++--- .../nosql/redis/scheduler/JobDispatcher.java | 67 ++++++++++++------- 4 files changed, 64 insertions(+), 38 deletions(-) diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/elector/MasterElectorDemo.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/elector/MasterElectorDemo.java index c6cd14d8..f11eb463 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/elector/MasterElectorDemo.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/elector/MasterElectorDemo.java @@ -13,9 +13,9 @@ public class MasterElectorDemo { JedisPool pool = JedisPoolFactory.createJedisPool(JedisUtils.DEFAULT_HOST, JedisUtils.DEFAULT_PORT, JedisUtils.DEFAULT_TIMEOUT, 1); try { - MasterElector masterElector = new MasterElector(pool, 10); + MasterElector masterElector = new MasterElector(pool, 5, 10); - masterElector.start(5); + masterElector.start(); System.out.println("Hit enter to stop."); while (true) { diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobDispatcherDemo.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobDispatcherDemo.java index e95c4ea0..98dcb407 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobDispatcherDemo.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobDispatcherDemo.java @@ -27,7 +27,7 @@ public class JobDispatcherDemo { printJobNumbers(jobDispatcher); - jobDispatcher.start(1000); + jobDispatcher.start(); System.out.println("Hit enter to stop."); while (true) { diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/elector/MasterElector.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/elector/MasterElector.java index 864cc34b..98f4c46f 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/elector/MasterElector.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/elector/MasterElector.java @@ -36,13 +36,14 @@ import redis.clients.jedis.JedisPool; */ public class MasterElector implements Runnable { - private static final String DEFAULT_MASTER_KEY = "master"; + public static final String DEFAULT_MASTER_KEY = "master"; private static Logger logger = LoggerFactory.getLogger(MasterElector.class); - private static final AtomicInteger poolNumber = new AtomicInteger(1); + private static AtomicInteger poolNumber = new AtomicInteger(1); private ScheduledExecutorService internalScheduledThreadPool; private ScheduledFuture electorJob; + private int intervalSeconds; private JedisTemplate jedisTemplate; @@ -51,11 +52,15 @@ public class MasterElector implements Runnable { private AtomicBoolean master = new AtomicBoolean(false); private String masterKey = DEFAULT_MASTER_KEY; - public MasterElector(JedisPool jedisPool, int expireSeconds) { + public MasterElector(JedisPool jedisPool, int intervalSeconds, int expireSeconds) { jedisTemplate = new JedisTemplate(jedisPool); this.expireSeconds = expireSeconds; + this.intervalSeconds = intervalSeconds; } + /** + * 发挥目前该实例是否master + */ public boolean isMaster() { return master.get(); } @@ -63,23 +68,23 @@ public class MasterElector implements Runnable { /** * 启动分发线程, 自行创建scheduler线程池. */ - public void start(long periodSeconds) { + public void start() { internalScheduledThreadPool = Executors.newScheduledThreadPool(1, Threads.buildJobFactory("Master-Elector-" + poolNumber.getAndIncrement() + "-%d")); - start(periodSeconds, internalScheduledThreadPool); + start(internalScheduledThreadPool); } /** * 启动分发线程, 使用传入的scheduler线程池. */ - public void start(long periodSeconds, ScheduledExecutorService scheduledThreadPool) { - if (periodSeconds >= expireSeconds) { + public void start(ScheduledExecutorService scheduledThreadPool) { + if (intervalSeconds >= expireSeconds) { throw new IllegalArgumentException("periodSeconds must less than expireSeconds. periodSeconds is " - + periodSeconds + " expireSeconds is " + expireSeconds); + + intervalSeconds + " expireSeconds is " + expireSeconds); } hostId = generateHostId(); - electorJob = scheduledThreadPool.scheduleAtFixedRate(new WrapExceptionRunnable(this), 0, periodSeconds, + electorJob = scheduledThreadPool.scheduleAtFixedRate(new WrapExceptionRunnable(this), 0, intervalSeconds, TimeUnit.SECONDS); logger.info("masterElector start, hostName:{}.", hostId); } @@ -95,6 +100,9 @@ public class MasterElector implements Runnable { } } + /** + * 生成host id的方法哦，可在子类重载. + */ protected String generateHostId() { String host = "localhost"; try { @@ -144,6 +152,9 @@ public class MasterElector implements Runnable { }); } + /** + * 如果应用中有多种master，设置唯一的master name + */ public void setMasterKey(String masterKey) { this.masterKey = masterKey; } 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 63fb5961..107f653c 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 @@ -30,28 +30,27 @@ import com.google.common.collect.Lists; */ public class JobDispatcher implements Runnable { public static final String DEFAULT_DISPATCH_LUA_FILE = "classpath:/redis/dispatch.lua"; + public static final long DEFAULT_INTERVAL_MILLIS = 1000; private static Logger logger = LoggerFactory.getLogger(JobDispatcher.class); - private static final AtomicInteger poolNumber = new AtomicInteger(1); + private static AtomicInteger poolNumber = new AtomicInteger(1); private ScheduledExecutorService internalScheduledThreadPool; + private ScheduledFuture dispatchJob; + private long intervalMillis = DEFAULT_INTERVAL_MILLIS; private JedisTemplate jedisTemplate; - private JedisScriptExecutor scriptExecutor; + private String scriptPath = DEFAULT_DISPATCH_LUA_FILE; private String scriptHash; private List keys; - private String readyJobKey; private String sleepingJobKey; + private String readyJobKey; private String dispatchCounterKey; public JobDispatcher(String jobName, JedisPool jedisPool) { - this(jobName, jedisPool, DEFAULT_DISPATCH_LUA_FILE); - } - - public JobDispatcher(String jobName, JedisPool jedisPool, String scriptPath) { sleepingJobKey = Keys.getSleepingJobKey(jobName); readyJobKey = Keys.getReadyJobKey(jobName); dispatchCounterKey = Keys.getDispatchCounterKey(jobName); @@ -59,10 +58,29 @@ public class JobDispatcher implements Runnable { jedisTemplate = new JedisTemplate(jedisPool); this.scriptExecutor = new JedisScriptExecutor(jedisPool); - loadLuaScript(scriptPath); } - private void loadLuaScript(String scriptPath) { + /** + * 启动分发线程, 自行创建scheduler线程池. + */ + public void start() { + internalScheduledThreadPool = Executors.newScheduledThreadPool(1, + Threads.buildJobFactory("Job-Dispatcher-" + poolNumber.getAndIncrement() + "-%d")); + + start(internalScheduledThreadPool); + } + + /** + * 启动分发线程, 使用传入的scheduler线程池. + */ + public void start(ScheduledExecutorService scheduledThreadPool) { + loadLuaScript(); + + dispatchJob = scheduledThreadPool.scheduleAtFixedRate(new WrapExceptionRunnable(this), 0, intervalMillis, + TimeUnit.MILLISECONDS); + } + + private void loadLuaScript() { String script; try { Resource resource = new DefaultResourceLoader().getResource(scriptPath); @@ -74,23 +92,6 @@ public class JobDispatcher implements Runnable { scriptHash = scriptExecutor.load(script); } - /** - * 启动分发线程, 自行创建scheduler线程池. - */ - public void start(long periodMillis) { - internalScheduledThreadPool = Executors.newScheduledThreadPool(1, - Threads.buildJobFactory("Job-Dispatcher-" + poolNumber.getAndIncrement() + "-%d")); - start(periodMillis, internalScheduledThreadPool); - } - - /** - * 启动分发线程, 使用传入的scheduler线程池. - */ - public void start(long periodMillis, ScheduledExecutorService scheduledThreadPool) { - dispatchJob = scheduledThreadPool.scheduleAtFixedRate(new WrapExceptionRunnable(this), 0, periodMillis, - TimeUnit.MILLISECONDS); - } - /** * 停止分发任务，如果是自行创建的threadPool则自行销毁。 */ @@ -140,4 +141,18 @@ public class JobDispatcher implements Runnable { public void restDispatchNumber() { jedisTemplate.set(dispatchCounterKey, "0"); } + + /** + * 设置非默认的script path, 格式为spring的Resource路径风格。 + */ + public void setScriptPath(String scriptPath) { + this.scriptPath = scriptPath; + } + + /** + * 设置非默认1秒的分发间隔. + */ + public void setIntervalMillis(long intervalMillis) { + this.intervalMillis = intervalMillis; + } }