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

继续改进redis相关代码及注释, consumer类型精简

Browse Source
This commit is contained in:
Calvin
2013-09-07 17:36:31 +08:00
parent 8506739bbe
commit 1c3be136cc
11 changed files with 212 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/BatchJobConsumerDemo.java → examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/AdvancedJobConsumerBatchPopDemo.java
+7 -7
View File
@@ -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);
}
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/ReliableJobConsumerDemo.java → examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/consumer/AdvancedJobConsumerSinglePopDemo.java
+6 -5
View File
@@ -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
modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/AdvancedConsumer.java
+145
View File
@@ -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<String> 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<String> 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<String> popupJobs() {
List<String> jobs = null;
try {
long currTime = System.currentTimeMillis();
List<String> args = Lists.newArrayList(String.valueOf(currTime), String.valueOf(batchSize),
String.valueOf(reliable));
jobs = (List<String>) 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;
}
}
modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/BatchJobConsumer.java
-99
View File
@@ -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<String> 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<String> popupJobs() {
List<String> jobs = null;
try {
long currTime = System.currentTimeMillis();
List<String> args = Lists.newArrayList(String.valueOf(currTime), String.valueOf(batchSize),
String.valueOf(reliable));
jobs = (List<String>) 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;
}
}
modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobProducer.java
+1 -13
View File
@@ -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) {
modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobStatistics.java
+6 -6
View File
@@ -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");
modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/ReliableJobConsumer.java
-84
View File
@@ -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<String> 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<String> 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;
}
}
modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/SimpleJobConsumer.java
+10 -4
View File
@@ -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<List<String>>() {
@Override
public List<String> 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;
}
}
modules/extension/src/main/resources/redis/batchpop.lua
+3 -3
View File
@@ -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
modules/extension/src/main/resources/redis/dispatch.lua
+32 -29
View File
@@ -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
modules/extension/src/main/resources/redis/reliablepop.lua → modules/extension/src/main/resources/redis/singlepop.lua
+2 -2
View File
@@ -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