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; + } }