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

#221 加入对Jedis的封装，注释与重命名

Browse Source
This commit is contained in:
Calvin
2013-07-11 01:44:09 +08:00
parent 327013719e
commit 4af636d29a
8 changed files with 46 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/JedisPoolFactory.java
+5 -5
View File
@@ -5,16 +5,16 @@ import redis.clients.jedis.JedisPoolConfig;
public class JedisPoolFactory {
public static JedisPool createJedisPool(String defaultHost, int defaultPort, int defaultTimeout, int poolSize) {
// merge default setting and system properties
public static JedisPool createJedisPool(String defaultHost, int defaultPort, int defaultTimeout, int threadCount) {
// 合并命令行传入的系统变量与默认值
String host = System.getProperty("benchmark.host", defaultHost);
String port = System.getProperty("benchmark.port", String.valueOf(defaultPort));
String timeout = System.getProperty("benchmark.timeout", String.valueOf(defaultTimeout));
// 设置Pool
// 设置Pool大小，设为与线程数等大。
JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxActive(poolSize);
poolConfig.setMaxIdle(poolSize);
poolConfig.setMaxActive(threadCount);
poolConfig.setMaxIdle(threadCount);
// create jedis pool
return new JedisPool(poolConfig, host, Integer.valueOf(port), Integer.valueOf(timeout));
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisCounterBenchmark.java
+5 -5
View File
@@ -23,13 +23,12 @@ public class RedisCounterBenchmark extends ConcurrentBenchmark {
private static final int DEFAULT_PORT = Protocol.DEFAULT_PORT;
private static final int DEFAULT_TIMEOUT = Protocol.DEFAULT_TIMEOUT;
private String counterName = "ss.counter";
private String counterKey = "ss.counter";
private JedisPool pool;
private JedisTemplate jedisTemplate;
public static void main(String[] args) throws Exception {
RedisCounterBenchmark benchmark = new RedisCounterBenchmark();
benchmark.execute();
}
@@ -41,7 +40,9 @@ public class RedisCounterBenchmark extends ConcurrentBenchmark {
protected void setUp() {
pool = JedisPoolFactory.createJedisPool(DEFAULT_HOST, DEFAULT_PORT, DEFAULT_TIMEOUT, threadCount);
jedisTemplate = new JedisTemplate(pool);
jedisTemplate.set(counterName, "0");
// 重置Counter
jedisTemplate.set(counterKey, "0");
}
@Override
@@ -58,8 +59,7 @@ public class RedisCounterBenchmark extends ConcurrentBenchmark {
@Override
protected void execute(int requestSequence) {
jedisTemplate.incr(counterName);
jedisTemplate.incr(counterKey);
}
}
}
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisMassInsertionBenchmark.java
+4 -3
View File
@@ -28,7 +28,6 @@ public class RedisMassInsertionBenchmark extends ConcurrentBenchmark {
private String keyPrefix = "ss.session:";
private int batchSize = 10;
private JsonMapper jsonMapper = new JsonMapper();
private JedisPool pool;
private JedisTemplate jedisTemplate;
@@ -44,11 +43,10 @@ public class RedisMassInsertionBenchmark extends ConcurrentBenchmark {
@Override
protected void setUp() {
// create jedis pool
pool = JedisPoolFactory.createJedisPool(DEFAULT_HOST, DEFAULT_PORT, DEFAULT_TIMEOUT, threadCount);
jedisTemplate = new JedisTemplate(pool);
// remove all keys
// 清空数据库
jedisTemplate.flushDB();
}
@@ -64,6 +62,9 @@ public class RedisMassInsertionBenchmark extends ConcurrentBenchmark {
public class MassInsertionTask extends BenchmarkTask {
/**
* 在最快速环境下, 循环在JedisAction中完成，所以没有使用父类的run框架。
*/
@Override
public void run() {
onThreadStart();
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisSessionBenchmark.java
+7 -5
View File
@@ -43,11 +43,10 @@ public class RedisSessionBenchmark extends ConcurrentBenchmark {
@Override
protected void setUp() {
// create jedis pool
pool = JedisPoolFactory.createJedisPool(DEFAULT_HOST, DEFAULT_PORT, DEFAULT_TIMEOUT, threadCount);
jedisTemplate = new JedisTemplate(pool);
// remove all keys
// 清空数据库
jedisTemplate.flushDB();
}
@@ -67,7 +66,7 @@ public class RedisSessionBenchmark extends ConcurrentBenchmark {
@Override
protected void execute(final int requestSequnce) {
final int randomIndex = random.nextInt((int) loopCount);
int randomIndex = random.nextInt((int) loopCount);
final String key = new StringBuilder().append(keyPrefix).append(taskSequence).append(":")
.append(randomIndex).toString();
@@ -77,10 +76,13 @@ public class RedisSessionBenchmark extends ConcurrentBenchmark {
Session session = new Session(key);
session.setAttrbute("name", key);
session.setAttrbute("seq", requestSequnce);
// set session expired after 300 seconds
session.setAttrbute("address", "address:" + requestSequnce);
session.setAttrbute("tel", "tel:" + requestSequnce);
// 设置session，超时时间为300秒
jedis.setex(session.getId(), 300, jsonMapper.toJson(session));
// also get it back
// 再重新从Redis中取出并反序列化
String sessionBackString = jedis.get(key);
Session sessionBack = jsonMapper.fromJson(sessionBackString, Session.class);
}
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobDispatcherDemo.java
+3 -2
View File
@@ -7,7 +7,9 @@ import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Protocol;
/**
* 每秒执行一次Lua Script，从ss.timer(Sorted Set)中取出到时的任务，放入ss.job(List)和ss.ack(Sorted Set)中等待领取和完成呢个确认.
* 运行JobDispatcher，每秒将Job从"ss.sleeping" sorted set 发布到"ss.ready" list.
*
* 可用系统参数重置相关变量，@see RedisCounterBenchmark
*
* @author calvin
*/
@@ -36,7 +38,6 @@ public class JobDispatcherDemo {
return;
}
}
} finally {
pool.destroy();
}
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobListenerDemo.java
+8 -4
View File
@@ -14,7 +14,9 @@ import redis.clients.jedis.JedisPool;
import com.google.common.util.concurrent.RateLimiter;
/**
* 消费者多线程执行BRPOP从ss.job(List)中阻塞取出Job，完成后执行ZREM从ss.ack(Sorted Set)中删除任务实现Ack.
* 多线程运行JobListener，从"ss.job:ready" list中popup job进行处理。
*
* 可用系统参数重置相关变量，@see RedisCounterBenchmark
*
* @author calvin
*/
@@ -67,12 +69,15 @@ public class JobListenerDemo implements JobHandler {
}
}
/**
* 处理Job的回调函数.
*/
@Override
public void receiveJob(String job) {
public void handleJob(String job) {
long globalCount = golbalCounter.incrementAndGet();
localCounter++;
// print global progress, 所有線程裡只有一個需要在10秒內打印
// print global progress, 所有線程裡只有一個线程会在10秒內打印一次。
if (golbalPrintRate.tryAcquire()) {
System.out.printf("Total pop %,d jobs, tps is %,d\n", globalCount,
(globalCount - golbalPreviousCount.get()) / PRINT_BETWEEN_SECONDS);
@@ -85,6 +90,5 @@ public class JobListenerDemo implements JobHandler {
(localCounter - localPreviousCount) / PRINT_BETWEEN_SECONDS);
localPreviousCount = localCounter;
}
}
}
examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobProducerDemo.java
+12 -11
View File
@@ -11,19 +11,20 @@ import org.springside.modules.test.benchmark.ConcurrentBenchmark;
import redis.clients.jedis.JedisPool;
/**
* 将Job放入ss.job:slepping(sorted set).
* 运行JobManager产生新的Job。
*
* 可用-Dbenchmark.thread.count, -Dbenchmark.loop.count 重置测试规模
* 可用-Dbenchmark.host,-Dbenchmark.port,-Dbenchmark.timeout 重置连接参数
* 可用系统参数重置相关变量，@see RedisCounterBenchmark
*
* @author calvin
*/
public class JobProducerDemo extends ConcurrentBenchmark {
private static final int DEFAULT_THREAD_COUNT = 5;
private static final long DEFAULT_LOOP_COUNT = 5000 * 60;
private static final long DEFAULT_LOOP_COUNT = 100000;
private static long delayInSeconds = JobDispatcherDemo.DELAY_SECONDS;
private static AtomicLong delayInSeconds = new AtomicLong(JobDispatcherDemo.DELAY_SECONDS);
private static AtomicLong idGenerator = new AtomicLong(0);
private long expectTps;
private JedisPool pool;
private JobManager jobManager;
@@ -34,11 +35,12 @@ public class JobProducerDemo extends ConcurrentBenchmark {
public JobProducerDemo() {
super(DEFAULT_THREAD_COUNT, DEFAULT_LOOP_COUNT);
this.expectTps = Long.parseLong(System.getProperty("benchmark.tps",
String.valueOf(JobDispatcherDemo.EXPECT_TPS)));
}
@Override
protected void setUp() {
// create jedis pool
pool = JedisPoolFactory.createJedisPool(JobDispatcherDemo.DEFAULT_HOST, JobDispatcherDemo.DEFAULT_PORT,
JobDispatcherDemo.DEFAULT_TIMEOUT, threadCount);
jobManager = new JobManager("ss", pool);
@@ -58,13 +60,12 @@ public class JobProducerDemo extends ConcurrentBenchmark {
@Override
public void execute(final int requestSequence) {
long jobId = idGenerator.getAndIncrement();
jobManager.scheduleJob("job:" + jobId, delayInSeconds, TimeUnit.SECONDS);
jobManager.scheduleJob("job:" + jobId, delayInSeconds.get(), TimeUnit.SECONDS);
// 达到TPS上限后，expireTime往后滚动一秒
if ((jobId % (JobDispatcherDemo.EXPECT_TPS)) == 0) {
delayInSeconds += 1;
// 达到期望的每秒的TPS后，expireTime往后滚动一秒
if ((jobId % (expectTps)) == 0) {
delayInSeconds.incrementAndGet();
}
}
}
}
modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobListener.java
+2 -2
View File
@@ -45,7 +45,7 @@ public class JobListener implements Runnable {
List<String> nameValuePair = jedis.brpop(POPUP_TIMEOUT, readyJobName);
if ((nameValuePair != null) && !nameValuePair.isEmpty()) {
String job = nameValuePair.get(1);
jobHandler.receiveJob(job);
jobHandler.handleJob(job);
}
}
}
@@ -53,6 +53,6 @@ public class JobListener implements Runnable {
}
public interface JobHandler {
void receiveJob(String job);
void handleJob(String job);
}
}