#221 加入对Jedis的封装,又一轮代码改进与注释,新增了JedisPoolConfig的设置函数

This commit is contained in:
Calvin
2013-07-14 00:34:59 +08:00
parent f8ef90f6b0
commit e70ff75ecd
12 changed files with 79 additions and 52 deletions
@@ -1,5 +1,7 @@
package org.springside.examples.showcase.demos.redis;
import org.springside.modules.nosql.redis.JedisUtils;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
@@ -11,10 +13,8 @@ public class JedisPoolFactory {
String port = System.getProperty("benchmark.port", String.valueOf(defaultPort));
String timeout = System.getProperty("benchmark.timeout", String.valueOf(defaultTimeout));
// 设置Pool大小,设为与线程数等大。
JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxActive(threadCount);
poolConfig.setMaxIdle(threadCount);
// 设置Pool大小,设为与线程数等大,并屏蔽掉idle checking
JedisPoolConfig poolConfig = JedisUtils.createPoolConfig(threadCount, threadCount);
// create jedis pool
return new JedisPool(poolConfig, host, Integer.valueOf(port), Integer.valueOf(timeout));
@@ -21,7 +21,7 @@ public class JobProducerDemo extends ConcurrentBenchmark {
private static final int DEFAULT_THREAD_COUNT = 5;
private static final long DEFAULT_LOOP_COUNT = 100000;
private static AtomicLong delayInSeconds = new AtomicLong(JobDispatcherDemo.DELAY_SECONDS);
private static AtomicLong delaySeconds = new AtomicLong(JobDispatcherDemo.DELAY_SECONDS);
private static AtomicLong idGenerator = new AtomicLong(0);
private long expectTps;
@@ -60,11 +60,11 @@ public class JobProducerDemo extends ConcurrentBenchmark {
@Override
public void execute(final int requestSequence) {
long jobId = idGenerator.getAndIncrement();
jobManager.scheduleJob("job:" + jobId, delayInSeconds.get(), TimeUnit.SECONDS);
jobManager.scheduleJob("job:" + jobId, delaySeconds.get(), TimeUnit.SECONDS);
// 达到期望的每秒的TPS后,expireTime往后滚动一秒
if ((jobId % (expectTps)) == 0) {
delayInSeconds.incrementAndGet();
delaySeconds.incrementAndGet();
}
}
}
@@ -58,7 +58,7 @@ public class JodaDemo {
// 将日期按UTC时区打印
DateTime zoneWithUTC = fooDate.withZone(DateTimeZone.UTC);
System.out.println(zoneWithUTC.toString(format) + " " + zoneWithUTC.getMillis());// "1978-06-01 04:10:08",
// sameMills
// sameMillis
// 按不同的时区分析字符串,得到不同的时间
String dateString = "1978-06-01 12:10:08";
@@ -12,7 +12,7 @@ public class StopWatch {
startTime = System.currentTimeMillis();
}
public long getMills() {
public long getMillis() {
return System.currentTimeMillis() - startTime;
}
@@ -70,7 +70,7 @@ public class JedisScriptExecutor {
@Override
public Object action(Jedis jedis) {
Object result = jedis.evalsha(hash, keys, args);
logger.debug("Script hash {} execution time is {}ms", hash, stopWatch.getMills());
logger.debug("Script hash {} execution time is {}ms", hash, stopWatch.getMillis());
return result;
}
});
@@ -78,7 +78,7 @@ public class JedisScriptExecutor {
logger.warn("Script hash {} is not loaded yet, try to reload and run it again", hash, e);
stopWatch.reset();
Object result = reloadAndExecute(hash, keys, args);
logger.debug("Script hash {} reload and execution time is {}ms", hash, stopWatch.getMills());
logger.debug("Script hash {} reload and execution time is {}ms", hash, stopWatch.getMillis());
return result;
}
}
@@ -193,6 +193,9 @@ public class JedisTemplate {
});
}
/**
* 返回List长度, key不存在时返回0,key类型不是list时抛出异常.
*/
public long llen(final String key) {
return execute(new JedisAction<Long>() {
@@ -203,13 +206,16 @@ public class JedisTemplate {
});
}
// ////////////// 关于Sorted Set ///////////////////////// //
public void zadd(final String key, final double score, final String member) {
execute(new JedisActionNoResult() {
// ////////////// 关于Sorted Set ///////////////////////////
/**
* 加入Sorted set, 如果member在Set里已存在,只更新score并返回false,否则返回true.
*/
public boolean zadd(final String key, final String member, final double score) {
return execute(new JedisAction<Boolean>() {
@Override
public void action(Jedis jedis) {
jedis.zadd(key, score, member);
public Boolean action(Jedis jedis) {
return jedis.zadd(key, score, member) == 1 ? true : false;
}
});
}
@@ -227,6 +233,9 @@ public class JedisTemplate {
});
}
/**
* 返回List长度, key不存在时返回0,key类型不是sorted set时抛出异常.
*/
public long zcard(final String key) {
return execute(new JedisAction<Long>() {
@@ -1,5 +1,7 @@
package org.springside.modules.nosql.redis;
import redis.clients.jedis.JedisPoolConfig;
public class JedisUtils {
private static final String OK_CODE = "OK";
private static final String OK_MULTI_CODE = "+OK";
@@ -10,4 +12,26 @@ public class JedisUtils {
public static boolean isStatusOk(String status) {
return (status != null) && (OK_CODE.equals(status) || OK_MULTI_CODE.equals(status));
}
/**
* 快速设置JedisPoolConfig, 不执行idle checking。
*/
public static JedisPoolConfig createPoolConfig(int maxIdle, int maxActive) {
JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxIdle(maxIdle);
poolConfig.setMaxActive(maxActive);
poolConfig.setTimeBetweenEvictionRunsMillis(-1);
return poolConfig;
}
/**
* 快速设置JedisPoolConfig, 设置执行idle checking的间隔和idle时间.
*/
public static JedisPoolConfig createPoolConfig(int maxIdle, int maxActive, int checkingIntervalSecs,
int idleTimeSecs) {
JedisPoolConfig poolConfig = createPoolConfig(maxIdle, maxActive);
poolConfig.setTimeBetweenEvictionRunsMillis(checkingIntervalSecs * 1000);
poolConfig.setMinEvictableIdleTimeMillis(idleTimeSecs * 1000);
return poolConfig;
}
}
@@ -34,23 +34,17 @@ public class JobDispatcher implements Runnable {
private static Logger logger = LoggerFactory.getLogger(JobDispatcher.class);
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private ScheduledExecutorService internalScheduledThreadPool;
private ScheduledFuture dispatchJob;
private JedisScriptExecutor scriptExecutor;
private JedisTemplate jedisTemplate;
private JedisScriptExecutor scriptExecutor;
private String scriptHash;
private List<String> keys;
private String readyJobKey;
private String sleepingJobKey;
private String dispatchCounterKey;
public JobDispatcher(String jobName, JedisPool jedisPool) {
@@ -83,17 +77,17 @@ public class JobDispatcher implements Runnable {
/**
* 启动分发线程, 自行创建scheduler线程池.
*/
public void start(long periodMilliseconds) {
public void start(long periodMillis) {
internalScheduledThreadPool = Executors.newScheduledThreadPool(1,
Threads.buildJobFactory("Job-Dispatcher-" + poolNumber.getAndIncrement() + "-%d"));
start(periodMilliseconds, internalScheduledThreadPool);
start(periodMillis, internalScheduledThreadPool);
}
/**
* 启动分发线程, 使用传入的scheduler线程池.
*/
public void start(long periodMilliseconds, ScheduledExecutorService scheduledThreadPool) {
dispatchJob = scheduledThreadPool.scheduleAtFixedRate(new WrapExceptionRunnable(this), 0, periodMilliseconds,
public void start(long periodMillis, ScheduledExecutorService scheduledThreadPool) {
dispatchJob = scheduledThreadPool.scheduleAtFixedRate(new WrapExceptionRunnable(this), 0, periodMillis,
TimeUnit.MILLISECONDS);
}
@@ -30,12 +30,12 @@ public class JobListener implements Runnable {
private static Logger logger = LoggerFactory.getLogger(JobListener.class);
private String readyJobKey;
private JedisTemplate jedisTemplate;
private final JobHandler jobHandler;
private String readyJobKey;
public JobListener(String jobName, JedisPool jedisPool, JobHandler jobHandler) {
jedisTemplate = new JedisTemplate(jedisPool);
readyJobKey = Keys.getReadyJobKey(jobName);
@@ -24,10 +24,10 @@ public class JobManager {
private static Logger logger = LoggerFactory.getLogger(JobManager.class);
private String sleepingJobKey;
private JedisTemplate jedisTemplate;
private String sleepingJobKey;
public JobManager(String jobName, JedisPool jedisPool) {
jedisTemplate = new JedisTemplate(jedisPool);
sleepingJobKey = Keys.getSleepingJobKey(jobName);
@@ -37,8 +37,8 @@ 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(sleepingJobKey, delayTimeInMillisecond, job);
final long delayTimeMillis = System.currentTimeMillis() + timeUnit.toMillis(delay);
jedisTemplate.zadd(sleepingJobKey, job, delayTimeMillis);
}
/**
@@ -56,7 +56,7 @@ public abstract class BenchmarkTask implements Runnable {
} catch (InterruptedException e) {
e.printStackTrace();
}
nextPrintTime = System.currentTimeMillis() + parent.intervalInMills;
nextPrintTime = System.currentTimeMillis() + parent.intervalMillis;
}
@@ -78,25 +78,25 @@ public abstract class BenchmarkTask implements Runnable {
long currentTime = System.currentTimeMillis();
if (currentTime > nextPrintTime) {
long lastIntervalInMills = parent.intervalInMills + (currentTime - nextPrintTime);
nextPrintTime = currentTime + parent.intervalInMills;
long lastIntervalMillis = parent.intervalMillis + (currentTime - nextPrintTime);
nextPrintTime = currentTime + parent.intervalMillis;
long lastRequests = currentRequests - previousRequests;
long totalTimeInMills = currentTime - parent.startTime.getTime();
long totalTimeInSeconds = TimeUnit.MILLISECONDS.toSeconds(totalTimeInMills);
long totalTimeMillis = currentTime - parent.startTime.getTime();
long totalTimeSeconds = TimeUnit.MILLISECONDS.toSeconds(totalTimeMillis);
long totalTps = (currentRequests * 1000) / totalTimeInMills;
long lastTps = (lastRequests * 1000) / lastIntervalInMills;
long totalTps = (currentRequests * 1000) / totalTimeMillis;
long lastTps = (lastRequests * 1000) / lastIntervalMillis;
BigDecimal lastLatency = new BigDecimal(lastIntervalInMills).divide(new BigDecimal(lastRequests), 2,
BigDecimal lastLatency = new BigDecimal(lastIntervalMillis).divide(new BigDecimal(lastRequests), 2,
BigDecimal.ROUND_HALF_UP);
BigDecimal totalLatency = new BigDecimal(totalTimeInMills).divide(new BigDecimal(currentRequests), 2,
BigDecimal totalLatency = new BigDecimal(totalTimeMillis).divide(new BigDecimal(currentRequests), 2,
BigDecimal.ROUND_HALF_UP);
System.out
.printf("Thread %02d process %,d requests after %s seconds. Last tps/latency is %,d/%sms. Total tps/latency is %,d/%sms.\n",
taskSequence, currentRequests, totalTimeInSeconds, lastTps, lastLatency.toString(),
taskSequence, currentRequests, totalTimeSeconds, lastTps, lastLatency.toString(),
totalTps, totalLatency.toString());
previousRequests = currentRequests;
@@ -107,10 +107,10 @@ public abstract class BenchmarkTask implements Runnable {
* 打印线程结果信息.
*/
protected void printThreadFinishMessage() {
long totalTimeInMills = System.currentTimeMillis() - parent.startTime.getTime();
long totalTimeMillis = System.currentTimeMillis() - parent.startTime.getTime();
long totalRequest = parent.loopCount;
long totalTps = (totalRequest * 1000) / totalTimeInMills;
BigDecimal totalLatency = new BigDecimal(totalTimeInMills).divide(new BigDecimal(totalRequest), 2,
long totalTps = (totalRequest * 1000) / totalTimeMillis;
BigDecimal totalLatency = new BigDecimal(totalTimeMillis).divide(new BigDecimal(totalRequest), 2,
BigDecimal.ROUND_HALF_UP);
System.out.printf("Thread %02d finish.Total tps/latency is %,d/%sms\n", taskSequence, totalTps,
@@ -23,7 +23,7 @@ public abstract class ConcurrentBenchmark {
protected CountDownLatch finishLock;
protected Date startTime;
protected int intervalInMills = 10 * 1000;
protected int intervalMillis = 10 * 1000;
public ConcurrentBenchmark(int defaultThreadCount, long defaultLoopCount) {
// merge default setting and system properties
@@ -80,15 +80,15 @@ public abstract class ConcurrentBenchmark {
Date endTime = new Date();
String className = this.getClass().getSimpleName();
long invokeTimes = threadCount * loopCount;
long timeInMills = endTime.getTime() - startTime.getTime();
long tps = (invokeTimes * 1000) / timeInMills;
long totalTimeMillis = endTime.getTime() - startTime.getTime();
long tps = (invokeTimes * 1000) / totalTimeMillis;
System.out.printf("%s finished at %s.\n%d threads processed %,d requests after %,d ms, tps is %,d.\n",
className, endTime.toString(), threadCount, invokeTimes, timeInMills, tps);
className, endTime.toString(), threadCount, invokeTimes, totalTimeMillis, tps);
}
protected void setIntervalInSecondss(int intervalInSeconds) {
this.intervalInMills = intervalInSeconds * 1000;
protected void setIntervalSeconds(int intervalSeconds) {
this.intervalMillis = intervalSeconds * 1000;
}
/**