#171 Redis增加session timer方案

examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisCounterBenchmark.java

@@ -14,15 +14,15 @@ import redis.clients.jedis.Protocol;
  * @author calvin
  */
 public class RedisCounterBenchmark extends ConcurrentBenchmark {
-	private static final int THREAD_COUNT = 20;
-	private static final long LOOP_COUNT = 50000;
+	private static final int THREAD_COUNT = 50;
+	private static final long LOOP_COUNT = 20000;
 	private static final int PRINT_BETWEEN_SECONDS = 10;
 
 	private static final String HOST = "localhost";
 	private static final int PORT = Protocol.DEFAULT_PORT;
 	private static final int TIMEOUT = Protocol.DEFAULT_TIMEOUT;
 
-	private String counterName = "springside.counter";
+	private String counterName = "ss.counter";
 	private JedisPool pool;
 
 	public static void main(String[] args) throws Exception {

examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisMassInsertionBenchmark.java

@@ -16,8 +16,8 @@ import redis.clients.jedis.Protocol;
  * @author calvin
  */
 public class RedisMassInsertionBenchmark extends ConcurrentBenchmark {
-	private static final int THREAD_COUNT = 20;
-	private static final long LOOP_COUNT = 500000;
+	private static final int THREAD_COUNT = 50;
+	private static final long LOOP_COUNT = 100000;
 	private static final int PRINT_BETWEEN_SECONDS = 10;
 	private static final int BATCH_SIZE = 10;
 
@@ -25,7 +25,7 @@ public class RedisMassInsertionBenchmark extends ConcurrentBenchmark {
 	private static final int PORT = Protocol.DEFAULT_PORT;
 	private static final int TIMEOUT = 40000;
 
-	private String keyPrefix = "springside.map:";
+	private String keyPrefix = "ss.map:";
 	private JsonMapper jsonMapper = new JsonMapper();
 	private JedisPool pool;
 

examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/RedisSessionBenchmark.java

@@ -17,15 +17,15 @@ import redis.clients.jedis.Protocol;
  * @author calvin
  */
 public class RedisSessionBenchmark extends ConcurrentBenchmark {
-	private static final int THREAD_COUNT = 20;
-	private static final long LOOP_COUNT = 50000;
+	private static final int THREAD_COUNT = 50;
+	private static final long LOOP_COUNT = 20000;
 	private static final int PRINT_BETWEEN_SECONDS = 10;
 
 	private static final String HOST = "localhost";
 	private static final int PORT = Protocol.DEFAULT_PORT;
 	private static final int TIMEOUT = Protocol.DEFAULT_TIMEOUT;
 
-	private String keyPrefix = "springside.map:";
+	private String keyPrefix = "ss.session:";
 	private JsonMapper jsonMapper = new JsonMapper();
 	private JedisPool pool;
 

examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/sessionTimer/RedisSessionTimerDistributor.java

@@ -0,0 +1,58 @@
+package org.springside.examples.showcase.demos.redis.sessionTimer;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.Protocol;
+
+import com.google.common.collect.Lists;
+
+/**
+ * 每秒执行一次Lua Script，从Sorted Set中取出到时间的任务，放入List中等待List中领取.
+ * 
+ * @author calvin
+ */
+public class RedisSessionTimerDistributor implements Runnable {
+
+	private static final String HOST = "localhost";
+	private static final int PORT = Protocol.DEFAULT_PORT;
+
+	private static String timerKey = "ss.timer";
+	private static String jobKey = "ss.job";
+	private static int batchSize = 2500;
+
+	private static Jedis jedis;
+	private int loop = 1;
+
+	public static void main(String[] args) throws Exception {
+		jedis = new Jedis(HOST, PORT);
+		jedis.connect();
+
+		RedisSessionTimerDistributor distributor = new RedisSessionTimerDistributor();
+		ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(1);
+		threadPool.scheduleAtFixedRate(distributor, 0, 1, TimeUnit.SECONDS);
+
+		System.out.println("Hit Any Key to stop");
+		System.in.read();
+		System.out.println("Shuting down");
+		threadPool.shutdownNow();
+		threadPool.awaitTermination(3, TimeUnit.SECONDS);
+
+		jedis.disconnect();
+	}
+
+	@Override
+	public void run() {
+		long startTime = System.currentTimeMillis();
+		//TODO: add to ack sorted set , and try to run multi key for lpush/zadd
+		String script = "local jobs=redis.call('zrangebyscore',KEYS[1],0,ARGV[1])\n"
+				+ " for i=1,ARGV[2] do \n  redis.call('lpush',KEYS[2],jobs[i])\n end\n"
+				+ " redis.call('zremrangebyscore',KEYS[1],0,ARGV[1])";
+		jedis.eval(script, Lists.newArrayList(timerKey, jobKey),
+				Lists.newArrayList(String.valueOf(loop * batchSize), String.valueOf(batchSize)));
+		loop++;
+		System.out.println("Times spend " + (System.currentTimeMillis() - startTime));
+	}
+}

examples/showcase/src/main/java/org/springside/examples/showcase/demos/redis/sessionTimer/RedisSessionTimerProducer.java

@@ -0,0 +1,87 @@
+package org.springside.examples.showcase.demos.redis.sessionTimer;
+
+import org.springside.modules.test.benchmark.BenchmarkTask;
+import org.springside.modules.test.benchmark.ConcurrentBenchmark;
+
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisPoolConfig;
+import redis.clients.jedis.Protocol;
+
+/**
+ * 将Timer插入sorted set.
+ * 
+ * @author calvin
+ */
+public class RedisSessionTimerProducer extends ConcurrentBenchmark {
+	private static final int THREAD_COUNT = 50;
+	private static final long LOOP_COUNT = 500 * 60;
+	private static final int PRINT_BETWEEN_SECONDS = 10;
+
+	private static final String HOST = "localhost";
+	private static final int PORT = Protocol.DEFAULT_PORT;
+	private static final int TIMEOUT = Protocol.DEFAULT_TIMEOUT;
+
+	private String keyPrefix = "ss.timer";
+	private JedisPool pool;
+
+	public static void main(String[] args) throws Exception {
+		RedisSessionTimerProducer benchmark = new RedisSessionTimerProducer(THREAD_COUNT, LOOP_COUNT);
+		benchmark.run();
+	}
+
+	public RedisSessionTimerProducer(int threadCount, long loopCount) {
+		super(threadCount, loopCount);
+	}
+
+	@Override
+	protected void onStart() {
+		//create jedis pool
+		JedisPoolConfig poolConfig = new JedisPoolConfig();
+		poolConfig.setMaxActive(THREAD_COUNT);
+		pool = new JedisPool(poolConfig, HOST, PORT, TIMEOUT);
+
+		//remove all keys
+		Jedis jedis = pool.getResource();
+		try {
+			jedis.flushDB();
+		} finally {
+			pool.returnResource(jedis);
+		}
+	}
+
+	@Override
+	protected void onFinish() {
+		pool.destroy();
+	}
+
+	@Override
+	protected BenchmarkTask createTask(int index) {
+		return new SessionTimerProducerTask(index, this, PRINT_BETWEEN_SECONDS);
+	}
+
+	public class SessionTimerProducerTask extends BenchmarkTask {
+
+		public SessionTimerProducerTask(int index, ConcurrentBenchmark parent, int printBetweenSeconds) {
+			super(index, parent, printBetweenSeconds);
+		}
+
+		@Override
+		public void run() {
+			Jedis jedis = pool.getResource();
+			onThreadStart();
+
+			try {
+
+				for (int i = 0; i < loopCount; i++) {
+					long scheduleTime = threadIndex * loopCount + i;
+					jedis.zadd(keyPrefix, scheduleTime, String.valueOf(scheduleTime));
+					printProgressMessage(i);
+				}
+			} finally {
+				onThreadFinish();
+				pool.returnResource(jedis);
+			}
+		}
+	}
+}