Redis Cluster多机房高可用实现--基于客户端

发表于   |   分类于   |     |   阅读次数 
1
本文以Redis-Cluster为例子，实际使用中Redis-Sentinel和Redis Standalone也是一样的。

一、现有问题

由于Redis本身的一些特性(例如复制)以及使用场景，造成Redis不太适合部署在不同的机房，所以通常来看Redis集群都是在同一个机房部署的。虽然Redis集群自身已经具备了高可用的特性，即使几个Redis节点异常或者挂掉，Redis Cluster也会实现故障自动转移，对应用方来说也可以在很短时间内恢复故障。但是如果发生了机房故障(断电、断网等极端情况)，如果应用方降级或者容错机制做的不好甚至业务本身不能降级，或者会丢失重要数据，或者可能瞬间会跑满应用的线程池造成服务不可用，对于一些重要的服务来说是非常致命的。

为了应对像机房故障这类情况，保证应用方在这种极端情况下，仍然可以正常服务(系统正常运行、数据正常)，所以需要给出一个Redis跨机房的方案。

二、实现思路和目标：

1.思路

  • 使用CacheCloud开通两个位于两个不同机房的Redis-Cluster集群(例如：兆维、北显)：一个叫major，作为主Redis服务，一个叫minor，作为备用Redis服务。
  • 开发定制版的客户端，利用netflix的hystrix组件能够解除依赖隔离的特性，在major出现故障时，将故障隔离，并将请求自动转发到minor，并且对于应用的主线程池没有影响。（有关hystrix的请求流程流程见下图，有关hystrix使用请参考：http://hot66hot.iteye.com/blog/2155036

2.实现目标：

  • 客户端易接入，如同使用Jedis API一样。
  • 真正实现跨机房的故障转移。
  • 依赖隔离，也就是说即使Redis出现问题，也不会影响主线程池。
  • 读取数据正常。
  • 写数据尽可能一致。
  • 更多的故障转移可配置参数(hystrix)：例如隔离线程池大小，超时等
  • 暴露相关统计数据和报表：如jmx和hystrix-dashboard

三、实施：

  • 1.利用hystrix能够隔离依赖的特性，为major和minor分别放到不同的线程池中(与应用的主线程池隔离)
  • 2.客户端接口和初始化方法：由于是定制化客户端，所以暂时没有通用的方法，所有的API需要自己实现。
1
2
3
4
5
6
7
public interface RedisCrossRoomClient {
    
    String set(String key, String value);
    
    String get(String key);
    
}

初始化方法，需要传入两个初始化好的PipeLineCluster

1
PipeLineCluster是我们内部对于JedisCluster的扩展，这里看成JedisCluster即可。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public class RedisClusterCrossRoomClientImpl implements RedisCrossRoomClient {
    private Logger logger = LoggerFactory.getLogger(RedisClusterCrossRoomClientImpl.class);
    /**
     * 主
     */
    private PipelineCluster majorPipelineCluster;
    /**
     * 备
     */
    private PipelineCluster minorPipelineCluster;
    
    public RedisClusterCrossRoomClientImpl(PipelineCluster majorPipelineCluster, PipelineCluster minorPipelineCluster) {
        this.majorPipelineCluster = majorPipelineCluster;
        this.minorPipelineCluster = minorPipelineCluster;
    }
}
  • 3.读操作方案：如下图，正常run指向到major, 异常(2.1图中所有指向getFallback)指向到minor。

例如：正常情况下都是从majorPipelineCluster读取数据，当出现非正常情况时（hystrix阀门开启、线程池拒绝、超时、异常）等情况时，走minorPipelineCluster的逻辑

基础类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public class BaseCommand {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    /**
     * hystrix参数,例如超时、线程池、关门策略、开门策略等等。
     */
    
    protected static final String MAJOR_READ_COMMAND_KEY = "major_read_command";
    protected static final String MAJOR_WRITE_COMMAND_KEY = "major_write_command";
    protected static final String MAJOR_GROUP_KEY = "major_redis_group";
    protected static final String MAJOR_THREAD_POOL_KEY = "major_redis_pool";
    public static int majorTimeOut = 1000;
    public static int majorThreads = 100;
    /**
     * hystrix参数,例如超时、线程池、关门策略、开门策略等等。
     */
    protected static final String MINOR_READ_COMMAND_KEY = "minor_read_command";
    protected static final String MINOR_WRITE_COMMAND_KEY = "minor_write_command";
    protected static final String MINOR_GROUP_KEY = "minor_redis_group";
    protected static final String MINOR_THREAD_POOL_KEY = "minor_redis_pool";
    public static int minorTimeOut = 1000;
    public static int minorThreads = 100;
}

读命令类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
public abstract class ReadCommand<T> extends BaseCommand {
    protected abstract T readMajor();
    protected abstract T readMinor();
    public T read() {
        // 1.收集总数
        RedisCrossRoomClientStatusCollector.collectCrossRoomStatus(HystrixStatCountTypeEnum.ALL);
        
        DataComponentCommand<T> majorCommand =
                new DataComponentCommand<T>(MAJOR_READ_COMMAND_KEY, MAJOR_GROUP_KEY, MAJOR_THREAD_POOL_KEY,
                        majorTimeOut, majorThreads) {
                    @Override
                    protected T run() throws Exception {
                        // 2.收集run总数
                        RedisCrossRoomClientStatusCollector.collectCrossRoomStatus(HystrixStatCountTypeEnum.RUN);
                        return readMajor();
                    }
                    @Override
                    public T getBusinessFallback() {
                        // 3.收集fallback总数
                        RedisCrossRoomClientStatusCollector.collectCrossRoomStatus(HystrixStatCountTypeEnum.FALLBACK_ALL);
                        
                        RedisCrossRoomHystrixStat.counterFallBack(MAJOR_READ_COMMAND_KEY);
                        return new DataComponentCommand<T>(MINOR_READ_COMMAND_KEY, MINOR_GROUP_KEY, MINOR_THREAD_POOL_KEY,
                                minorTimeOut, minorThreads) {
                            @Override
                            protected T run() throws Exception {
                                // 4.收集fallback-run总数
                                RedisCrossRoomClientStatusCollector.collectCrossRoomStatus(HystrixStatCountTypeEnum.FALLBACK_RUN);
                                return readMinor();
                            }
                            @Override
                            public T getBusinessFallback() throws RedisCrossRoomReadMinorFallbackException {
                                // 5.收集fallback-fallback总数
                                RedisCrossRoomClientStatusCollector.collectCrossRoomStatus(HystrixStatCountTypeEnum.FALLBACK_FALLBACK);
                                throw new RedisCrossRoomReadMinorFallbackException("MinorFallbackException");
                            }
                        }.execute();
                    }
                };
        return majorCommand.execute();
    }
}

例如get(String key)命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public class RedisClusterCrossRoomClientImpl implements RedisCrossRoomClient {
...
	@Override
	public String get(final String key) {
	    return new ReadCommand<String>() {
	
	        @Override
	        protected String readMajor() {
	            return majorPipelineCluster.get(key);
	        }
	
	        @Override
	        protected String readMinor() {
	            return minorPipelineCluster.get(key);
	        }
	    }.read();
	
	}
...
}
  • 4.写操作方案目标：尽可能双写，如果发生故障暂时只是做了隔离，没有做数据同步处理（未来会考虑接入MQ），目前只把写入的结果返回给应用方，应用方来维持一致性。

MultiWriteResult类，四个成员变量分别为：

序号 参数 含义
1 DataStatusEnum majorStatus 主集群执行结果状态
2 T majorResult 主集群执行Redis命令结果
3 DataStatusEnum minorStatus 备用集群执行结果状态
4 T minorResult 备用集群执行Redis命令结果 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
public abstract class WriteCommand<T> extends BaseCommand {
    protected abstract T writeMajor();
    protected abstract T writeMinor();
    
    protected abstract String getCommandParam();
    public MultiWriteResult<T> write() {
        DataComponentCommand<T> majorCommand =
                new DataComponentCommand<T>(MAJOR_WRITE_COMMAND_KEY, MAJOR_GROUP_KEY, MAJOR_THREAD_POOL_KEY,
                        majorTimeOut, majorThreads) {
                    @Override
                    protected T run() throws Exception {
                        return writeMajor();
                    }
                    @Override
                    public T getBusinessFallback() {
                        logger.warn("major cross-room failed: {}", getCommandParam());
                        return null;
                    }
                };
        DataComponentCommand<T> minorCommand =
                new DataComponentCommand<T>(MINOR_WRITE_COMMAND_KEY, MINOR_GROUP_KEY, MINOR_THREAD_POOL_KEY,
                        minorTimeOut, minorThreads) {
                    @Override
                    protected T run() throws Exception {
                        return writeMinor();
                    }
                    @Override
                    public T getBusinessFallback() {
                        logger.warn("minor cross-room failed: {}", getCommandParam());
                        return null;
                    }
                };
        Future<T> majorFuture = majorCommand.queue();
        Future<T> minorFuture = minorCommand.queue();
        T majorResult = null;
        T minorResult = null;
        try {
            majorResult = majorFuture.get();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        try {
            minorResult = minorFuture.get();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        DataStatusEnum majorStatus = DataStatusEnum.SUCCESS;
        DataStatusEnum minorStatus = DataStatusEnum.SUCCESS;
        if (majorResult == null) {
            majorStatus = DataStatusEnum.FAIL;
        }
        if (minorResult == null) {
            minorStatus = DataStatusEnum.FAIL;
        }
        return new MultiWriteResult<T>(majorStatus, majorResult, minorStatus, minorResult);
    }
}

例如set命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class RedisClusterCrossRoomClientImpl implements RedisCrossRoomClient {
...
	@Override
	public MultiWriteResult<String> set(final String key, final String value) {
	    return new WriteCommand<String>() {
	        @Override
	        protected String writeMajor() {
	            return majorPipelineCluster.set(key, value);
	        }
	
	        @Override
	        protected String writeMinor() {
	            return minorPipelineCluster.set(key, value);
	        }
	
	        @Override
	        protected String getCommandParam() {
	            return String.format("set key %s value %s", key, value);
	        }
	    }.write();
	}
...

四、对外暴露的数据和报表：

(1) hystrix-dashboard报表：实时统计图。
(2) jmx相关数据：major和minor相关统计，run和fallback调用次数、异常次数。

五、测试读：

1.major服务正常，但是major的线程池确实不够用

(1) 测试代码

测试方法：major的线程池设置小一些，请求的并发量大一些，每个线程做1000次随机读并返回主线程
测试验证：每个请求都有返回结果(前提是key是存在的)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
/**
 * 主线程池跑满：线程池size过小(major:30,minor:80，并发请求线程100)
 * 
 * @throws InterruptedException
 */
@Test
public void testRandomReadWithEnoughThreads() throws InterruptedException {
    redisClusterCrossRoomClient.setMajorThreads(30);
    redisClusterCrossRoomClient.setMinorThreads(80);
    int threadNum = 100;
    int perSize = TOTAL_SIZE / threadNum;
    int totalNeedGet = 1000;
    CountDownLatch countDownLatch = new CountDownLatch(threadNum);
    for (int i = 0; i < threadNum; i++) {
        int start = perSize * i + 1;
        int end = perSize * (i + 1);
        Thread thread = new RandomReadThread(start, end, totalNeedGet, countDownLatch);
        thread.start();
    }
    countDownLatch.await();
    System.out.println("request counter: " + TOTAL_SIZE);
    System.out.println("readSuccess counter:" + readSuccessCounter.get());
}
class RandomReadThread extends Thread {
    private int start;
    private int end;
    private int totalNeedGet;
    private CountDownLatch countDownLatch;
    private long counter;
    public RandomReadThread(int start, int end, int totalNeedGet, CountDownLatch countDownLatch) {
        this.start = start;
        this.end = end;
        this.totalNeedGet = totalNeedGet;
        this.countDownLatch = countDownLatch;
    }
    @Override
    public void run() {
        while (true) {
            try {
                if (counter >= totalNeedGet) {
                    countDownLatch.countDown();
                    break;
                }
                if (counter % 100 == 0) {
                    logger.info("{} execute {} th, total size {}", Thread.currentThread().getName(), counter,
                            totalNeedGet);
                }
                int id = start + new Random().nextInt(end - start);
                String key = "user:" + id;
                String result = redisClusterCrossRoomClient.get(key);
                if (StringUtils.isBlank(result)) {
                    logger.warn("key {}, value is null", key);
                } else {
                    readSuccessCounter.incrementAndGet();
                }
                counter++;
                TimeUnit.MILLISECONDS.sleep(10);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
(2) 故障转移：

major线程池偶尔吃满，将线程拒绝，并执行降级逻辑，将请求自动转移到minor。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
2016-04-25 15:07:10,658 ERROR [Thread-186] impl.RedisClusterCrossRoomClientImpl: major_redis_command could not be queued for execution and failed retrieving fallback.
com.netflix.hystrix.exception.HystrixRuntimeException: major_redis_command could not be queued for execution and failed retrieving fallback.
	at com.netflix.hystrix.HystrixCommand.getFallbackOrThrowException(HystrixCommand.java:1660)
	at com.netflix.hystrix.HystrixCommand.subscribeWithThreadIsolation(HystrixCommand.java:1250)
	at com.netflix.hystrix.HystrixCommand.access$1400(HystrixCommand.java:103)
	at com.netflix.hystrix.HystrixCommand$2.call(HystrixCommand.java:829)
	at com.netflix.hystrix.HystrixCommand$2.call(HystrixCommand.java:803)
	at rx.Observable.subscribe(Observable.java:6178)
	at com.netflix.hystrix.HystrixCommand$TimeoutObservable$1.call(HystrixCommand.java:1080)
	at com.netflix.hystrix.HystrixCommand$TimeoutObservable$1.call(HystrixCommand.java:1007)
	at rx.Observable$2.call(Observable.java:153)
	at rx.Observable$2.call(Observable.java:149)
	at rx.Observable$2.call(Observable.java:153)
	at rx.Observable$2.call(Observable.java:149)
	at rx.Observable.subscribe(Observable.java:6178)
	at com.netflix.hystrix.HystrixCommand$ObservableCommand$1.call(HystrixCommand.java:936)
	at com.netflix.hystrix.HystrixCommand$ObservableCommand$1.call(HystrixCommand.java:932)
	at rx.Observable.subscribe(Observable.java:6178)
	at rx.operators.BlockingOperatorToFuture.toFuture(BlockingOperatorToFuture.java:55)
	at rx.observables.BlockingObservable.toFuture(BlockingObservable.java:430)
	at com.netflix.hystrix.HystrixCommand.queue(HystrixCommand.java:477)
	at com.sohu.tv.cachecloud.client.redis.crossroom.impl.RedisClusterCrossRoomClientImpl.get(RedisClusterCrossRoomClientImpl.java:136)
	at com.sohu.tv.cachecloud.client.redis.crossroom.RedisClusterCrossRoomClientImplTest$RandomReadThread.run(RedisClusterCrossRoomClientImplTest.java:212)
Caused by: java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@5a56156e rejected from java.util.concurrent.ThreadPoolExecutor@656a0adc[Running, pool size = 30, active threads = 30, queued tasks = 0, completed tasks = 410]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2048)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:821)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1372)
	at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:132)
	at com.netflix.hystrix.HystrixCommand.subscribeWithThreadIsolation(HystrixCommand.java:1166)
	... 19 more
(3) hystrix-dashbord: 部分被major线程池拒绝的线程(紫色)，通过fallback转移到minor线程池中执行。

(4) 最终结果，返回非空的结果的个数等于请求个数
1
2
2016-04-25 15:08:43,393 INFO [main] crossroom.RedisClusterCrossRoomClientImplTest: request counter: 100000
2016-04-25 15:08:43,393 INFO [main] crossroom.RedisClusterCrossRoomClientImplTest: readSuccess counter: 100000

2. major服务异常，造成major的线程池不够用，或者存在大量异常，大量超时等等

(1) 测试代码：

测试方法：直接利用Redis的debug sleep seconds命令使得Redis暂时不提供服务。
测试验证：每个请求都有返回结果(前提是key是存在的)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
/**
 * 主线程池跑满：主服务异常(用redis sleep进行模拟)
 * 
 * @throws InterruptedException
 */
@Test
public void testRandomReadWithMajorSleep() throws InterruptedException {
    //让major下的所有节点休息10秒
    Map<String, JedisPool> jedisPoolMap = majorPipelineCluster.getConnectionHandler().getNodes();
    for (JedisPool jedisPool : jedisPoolMap.values()) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            jedis.debug(DebugParams.SLEEP(10));
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
    }
    int threadNum = 20;
    int perSize = TOTAL_SIZE / threadNum;
    int totalNeedGet = 1000;
    CountDownLatch countDownLatch = new CountDownLatch(threadNum);
    for (int i = 0; i < threadNum; i++) {
        int start = perSize * i + 1;
        int end = perSize * (i + 1);
        Thread thread = new RandomReadThread(start, end, totalNeedGet, countDownLatch);
        thread.start();
    }
    countDownLatch.await();
    logger.info("request counter: {}", (threadNum * totalNeedGet));
    logger.info("readSuccess counter: {}", readSuccessCounter.get());
}
(2) 故障转移：

直接major线程池的阀门打开了，所有请求执行降级逻辑，将请求自动转移到minor。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
2016-04-25 14:41:48,144 ERROR [Thread-13] impl.RedisClusterCrossRoomClientImpl: major_redis_command short-circuited and failed retrieving fallback.
com.netflix.hystrix.exception.HystrixRuntimeException: major_redis_command short-circuited and failed retrieving fallback.
 at com.netflix.hystrix.HystrixCommand.getFallbackOrThrowException(HystrixCommand.java:1660)
 at com.netflix.hystrix.HystrixCommand.getFallbackOrThrowException(HystrixCommand.java:1614)
 at com.netflix.hystrix.HystrixCommand.access$1300(HystrixCommand.java:103)
 at com.netflix.hystrix.HystrixCommand$2.call(HystrixCommand.java:820)
 at com.netflix.hystrix.HystrixCommand$2.call(HystrixCommand.java:803)
 at rx.Observable.subscribe(Observable.java:6178)
 at com.netflix.hystrix.HystrixCommand$TimeoutObservable$1.call(HystrixCommand.java:1080)
 at com.netflix.hystrix.HystrixCommand$TimeoutObservable$1.call(HystrixCommand.java:1007)
 at rx.Observable$2.call(Observable.java:153)
 at rx.Observable$2.call(Observable.java:149)
 at rx.Observable$2.call(Observable.java:153)
 at rx.Observable$2.call(Observable.java:149)
 at rx.Observable.subscribe(Observable.java:6178)
 at com.netflix.hystrix.HystrixCommand$ObservableCommand$1.call(HystrixCommand.java:936)
 at com.netflix.hystrix.HystrixCommand$ObservableCommand$1.call(HystrixCommand.java:932)
 at rx.Observable.subscribe(Observable.java:6178)
 at rx.operators.BlockingOperatorToFuture.toFuture(BlockingOperatorToFuture.java:55)
 at rx.observables.BlockingObservable.toFuture(BlockingObservable.java:430)
 at com.netflix.hystrix.HystrixCommand.queue(HystrixCommand.java:477)
 at com.sohu.tv.cachecloud.client.redis.crossroom.impl.RedisClusterCrossRoomClientImpl.get(RedisClusterCrossRoomClientImpl.java:136)
 at com.sohu.tv.cachecloud.client.redis.crossroom.RedisClusterCrossRoomClientImplTest$RandomReadThread.run(RedisClusterCrossRoomClientImplTest.java:214)
2016-04-25 14:41:48,145 ERROR [Thread-25] impl.RedisClusterCrossRoomClientImpl: major_redis_command short-circuited and failed retrieving fallback.
com.netflix.hystrix.exception.HystrixRuntimeException: major_redis_command short-circuited and failed retrieving fallback.
 at com.netflix.hystrix.HystrixCommand.getFallbackOrThrowException(HystrixCommand.java:1660)
 at com.netflix.hystrix.HystrixCommand.getFallbackOrThrowException(HystrixCommand.java:1614)
 at com.netflix.hystrix.HystrixCommand.access$1300(HystrixCommand.java:103)
 at com.netflix.hystrix.HystrixCommand$2.call(HystrixCommand.java:820)
 at com.netflix.hystrix.HystrixCommand$2.call(HystrixCommand.java:803)
 at rx.Observable.subscribe(Observable.java:6178)
 at com.netflix.hystrix.HystrixCommand$TimeoutObservable$1.call(HystrixCommand.java:1080)
 at com.netflix.hystrix.HystrixCommand$TimeoutObservable$1.call(HystrixCommand.java:1007)
 at rx.Observable$2.call(Observable.java:153)
 at rx.Observable$2.call(Observable.java:149)
 at rx.Observable$2.call(Observable.java:153)
 at rx.Observable$2.call(Observable.java:149)
 at rx.Observable.subscribe(Observable.java:6178)
 at com.netflix.hystrix.HystrixCommand$ObservableCommand$1.call(HystrixCommand.java:936)
 at com.netflix.hystrix.HystrixCommand$ObservableCommand$1.call(HystrixCommand.java:932)
 at rx.Observable.subscribe(Observable.java:6178)
 at rx.operators.BlockingOperatorToFuture.toFuture(BlockingOperatorToFuture.java:55)
 at rx.observables.BlockingObservable.toFuture(BlockingObservable.java:430)
 at com.netflix.hystrix.HystrixCommand.queue(HystrixCommand.java:477)
 at com.sohu.tv.cachecloud.client.redis.crossroom.impl.RedisClusterCrossRoomClientImpl.get(RedisClusterCrossRoomClientImpl.java:136)
 at com.sohu.tv.cachecloud.client.redis.crossroom.RedisClusterCrossRoomClientImplTest$RandomReadThread.run(RedisClusterCrossRoomClientImplTest.java:214)
(3) hystrix-dashboard

major的熔断器阀门已经打开，请求都转移到迁移到minor，之后major恢复，流量又回到major

(4) 最终结果：

返回非空的结果的个数等于请求个数

1
2
2016-04-25 15:01:21,338 INFO  [main] crossroom.RedisClusterCrossRoomClientImplTest: request counter: 20000
2016-04-25 15:01:21,338 INFO  [main] crossroom.RedisClusterCrossRoomClientImplTest: readSuccess counter: 20000

六、测试写：

1.major异常：

(1)测试代码

直接sleep major，20个线程，一共写1000次，每个线程50个key 。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
	@Test
public void testRandomWriteWithMajorSleep() throws InterruptedException {
    // major 休息10秒
    tempSleepPipelineCluster(majorPipelineCluster, 10);
    int totalWrite = 1000;
    int threadNum = 20;
    int perSize = totalWrite / threadNum;
    CountDownLatch countDownLatch = new CountDownLatch(threadNum);
    for (int i = 0; i < threadNum; i++) {
        int start = TOTAL_SIZE + perSize * i + 1;
        int end = TOTAL_SIZE + perSize * (i + 1);
        Thread thread = new RandomWriteThread(start, end, countDownLatch);
        thread.start();
    }
    countDownLatch.await();
    logger.info("total request write: {}", totalWrite);
    logger.info("writeFail counter: {}", writeFailCounter.get());
    logger.info("writePartFail counter: {}", writePartFailCounter.get());
    logger.info("writeSuccess counter: {}", writeSuccessCounter.get());
}
/**
 * 随机写
 * 
 * @author leifu
 * @Date 2016年4月25日
 * @Time 下午3:39:13
 */
class RandomWriteThread extends Thread {
    private int start;
    private int end;
    private CountDownLatch countDownLatch;
    public RandomWriteThread(int start, int end, CountDownLatch countDownLatch) {
        this.start = start;
        this.end = end;
        this.countDownLatch = countDownLatch;
    }
    @Override
    public void run() {
        for (int id = start; id <= end; id++) {
            try {
                String key = "user:" + id;
                String value = String.valueOf(id);
                MultiWriteResult<String> multiWriteResult = redisClusterCrossRoomClient.set(key, value);
                DataStatusEnum majorStatus = multiWriteResult.getMajorStatus();
                DataStatusEnum minorStatus = multiWriteResult.getMinorStatus();
                if(DataStatusEnum.SUCCESS.equals(majorStatus) && DataStatusEnum.SUCCESS.equals(minorStatus)){
                    writeSuccessCounter.incrementAndGet();
                } else if(!DataStatusEnum.SUCCESS.equals(majorStatus) && !DataStatusEnum.SUCCESS.equals(minorStatus)){
                    writeFailCounter.incrementAndGet();
                } else {
                    writePartFailCounter.incrementAndGet();
                }
                
                
                TimeUnit.MILLISECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        countDownLatch.countDown();
    }
}
(2) 异常：major出现大量超时，major写入失败
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
2016-04-25 17:31:08,703 ERROR [main] crossroom.RedisClusterCrossRoomClientImplTest: java.net.SocketTimeoutException: Read timed out
redis.clients.jedis.exceptions.JedisConnectionException: java.net.SocketTimeoutException: Read timed out
 at redis.clients.util.RedisInputStream.ensureFill(RedisInputStream.java:201)
 at redis.clients.util.RedisInputStream.readByte(RedisInputStream.java:40)
 at redis.clients.jedis.Protocol.process(Protocol.java:142)
 at redis.clients.jedis.Protocol.read(Protocol.java:206)
 at redis.clients.jedis.Connection.readProtocolWithCheckingBroken(Connection.java:282)
 at redis.clients.jedis.Connection.getStatusCodeReply(Connection.java:208)
 at redis.clients.jedis.BinaryJedis.debug(BinaryJedis.java:2974)
 at com.sohu.tv.cachecloud.client.redis.crossroom.RedisClusterCrossRoomClientImplTest.tempSleepPipelineCluster(RedisClusterCrossRoomClientImplTest.java:184)
 at com.sohu.tv.cachecloud.client.redis.crossroom.RedisClusterCrossRoomClientImplTest.testRandomWriteWithMajorSleep(RedisClusterCrossRoomClientImplTest.java:252)
 at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
 at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
 at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
 at java.lang.reflect.Method.invoke(Method.java:606)
 at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:47)
 at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
 at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:44)
 at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
 at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:271)
 at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:70)
 at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:50)
 at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238)
 at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63)
 at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236)
 at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53)
 at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:229)
 at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
 at org.junit.runners.ParentRunner.run(ParentRunner.java:309)
 at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:50)
 at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38)
 at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:467)
 at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:683)
 at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:390)
 at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:197)
Caused by: java.net.SocketTimeoutException: Read timed out
 at java.net.SocketInputStream.socketRead0(Native Method)
 at java.net.SocketInputStream.read(SocketInputStream.java:152)
 at java.net.SocketInputStream.read(SocketInputStream.java:122)
 at java.net.SocketInputStream.read(SocketInputStream.java:108)
 at redis.clients.util.RedisInputStream.ensureFill(RedisInputStream.java:195)
 ... 32 more
2016-04-25 17:31:10,777 WARN [Thread-14] sources.URLConfigurationSource: No URLs will be polled as dynamic configuration sources.
2016-04-25 17:31:10,777 INFO [Thread-14] sources.URLConfigurationSource: To enable URLs as dynamic configuration sources, define System property archaius.configurationSource.additionalUrls or make config.properties available on classpath.
2016-04-25 17:31:10,779 INFO [Thread-14] config.DynamicPropertyFactory: DynamicPropertyFactory is initialized with configuration sources: com.netflix.config.ConcurrentCompositeConfiguration@224dc69c
(3) dashboard:

(4) 最终结果：

一共写了1000次，成功写入22次，部分成功写入了978次。

1
2
3
4
5
6
7
2016-04-25 17:31:12,399 INFO [main] crossroom.RedisClusterCrossRoomClientImplTest: total request write: 1000
2016-04-25 17:31:12,399 INFO [main] crossroom.RedisClusterCrossRoomClientImplTest: writeFail counter: 0
2016-04-25 17:31:12,399 INFO [main] crossroom.RedisClusterCrossRoomClientImplTest: writePartFail counter: 978
2016-04-25 17:31:12,399 INFO [main] crossroom.RedisClusterCrossRoomClientImplTest: writeSuccess counter: 22
期间打印了失败的日志：
2016-04-28 10:47:29,871 WARN [Thread-17] impl.RedisClusterCrossRoomClientImpl$1: major cross-room failed: set key user:10300 value 10300
...................................

七、问题和展望

  • 由于是定制化客户端，目前只支持部分Redis命令API，后续需要添加
  • 在出现故障时，读正常，但是双写会出现major和minor数据不一致的情况。（可以考虑利用MQ机制来解决这个问题），目前使用打印日志的形式记录。
  • …………..