HashConcurrencyTest.java

package redis.netty.client;

import org.junit.Test;
import redis.netty.MultiBulkReply;
import spullara.util.functions.Block;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.junit.Assert.fail;

/**
 * Test a high concurrent application accessing a large hash table in redis.
 */
public class HashConcurrencyTest {
  @Test
  public void testConcurrency() throws ExecutionException, InterruptedException {
    RedisClient client = RedisClient.connect("localhost", 6379).get();
    for (int i = 0; i < 100; i++) {
      client.hset("hash", "key" + i, "values" + i).get();
    }
    long start = System.currentTimeMillis();
    final AtomicBoolean failed = new AtomicBoolean();
    final Semaphore semaphore = new Semaphore(100);
    for (int i = 0; i < 1000000; i++) {
      semaphore.acquireUninterruptibly();
      client.hgetall("hash").onSuccess(new Block<MultiBulkReply>() {
        @Override
        public void apply(MultiBulkReply multiBulkReply) {
          semaphore.release();
        }
      }).onFailure(new Block<Throwable>() {
        @Override
        public void apply(Throwable throwable) {
          throwable.printStackTrace();
          semaphore.release();
          failed.set(true);
        }
      });
      if (failed.get()) {
        fail("Concurrency bug");
        return;
      }
    }
    semaphore.acquireUninterruptibly(100);
    System.out.println(System.currentTimeMillis() - start);
  }

}