Alawn/springside4
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

#221 加入对Jedis的封装，改变函数定义更方便Spring注入

Browse Source
This commit is contained in:
Calvin
2013-07-14 23:35:49 +08:00
parent 3720c3555c
commit 7bbf836fbf
4 changed files with 64 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/elector/MasterElectorDemo.java
+2 -2
View File
@@ -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) {
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobDispatcherDemo.java
+1 -1
View File
@@ -27,7 +27,7 @@ public class JobDispatcherDemo {
printJobNumbers(jobDispatcher);
jobDispatcher.start(1000);
jobDispatcher.start();
System.out.println("Hit enter to stop.");
while (true) {
modules/extension/src/main/java/org/springside/modules/nosql/redis/elector/MasterElector.java
+20 -9
View File
@@ -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;
}
modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobDispatcher.java
+41 -26
View File
@@ -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<String> 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;
}
}