From 4af636d29aabc0dacfcffa86083bb0c35a690d55 Mon Sep 17 00:00:00 2001 From: Calvin Date: Thu, 11 Jul 2013 01:44:09 +0800 Subject: [PATCH] =?UTF-8?q?#221=20=E5=8A=A0=E5=85=A5=E5=AF=B9Jedis?= =?UTF-8?q?=E7=9A=84=E5=B0=81=E8=A3=85=EF=BC=8C=E6=B3=A8=E9=87=8A=E4=B8=8E?= =?UTF-8?q?=E9=87=8D=E5=91=BD=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../demos/redis/JedisPoolFactory.java | 10 ++++---- .../demos/redis/RedisCounterBenchmark.java | 10 ++++---- .../redis/RedisMassInsertionBenchmark.java | 7 +++--- .../demos/redis/RedisSessionBenchmark.java | 12 ++++++---- .../demos/redis/job/JobDispatcherDemo.java | 5 ++-- .../demos/redis/job/JobListenerDemo.java | 12 ++++++---- .../demos/redis/job/JobProducerDemo.java | 23 ++++++++++--------- .../nosql/redis/scheduler/JobListener.java | 4 ++-- 8 files changed, 46 insertions(+), 37 deletions(-) diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/JedisPoolFactory.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/JedisPoolFactory.java index 1f50af21..7a2f86c4 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/JedisPoolFactory.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/JedisPoolFactory.java @@ -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)); diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisCounterBenchmark.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisCounterBenchmark.java index d27df8cc..1b03e270 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisCounterBenchmark.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisCounterBenchmark.java @@ -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); } } } diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisMassInsertionBenchmark.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisMassInsertionBenchmark.java index 88029cde..1bfb0f6e 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisMassInsertionBenchmark.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisMassInsertionBenchmark.java @@ -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(); diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisSessionBenchmark.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisSessionBenchmark.java index 5afae738..06f13765 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisSessionBenchmark.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisSessionBenchmark.java @@ -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); } 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 91822d65..b503286a 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 @@ -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(); } diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobListenerDemo.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobListenerDemo.java index e009cf0e..8fb42d5c 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobListenerDemo.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobListenerDemo.java @@ -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; } - } } diff --git a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobProducerDemo.java b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobProducerDemo.java index d1b03fc1..fc6748e4 100644 --- a/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobProducerDemo.java +++ b/examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/job/JobProducerDemo.java @@ -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(); } - } } } diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobListener.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobListener.java index 2c8d4797..b2df331a 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobListener.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/scheduler/JobListener.java @@ -45,7 +45,7 @@ public class JobListener implements Runnable { List 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); } }