首页 话题 library logo 文库 我的
更多
回到详情

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

出处：https://cachecloud.github.io/2016/11/03/Redis%20Cluster%E5%A4%9A%E6%9C%BA%E6%88%BF%E9%AB%98%E5%8F%AF%E7%94%A8%E5%AE%9E%E7%8E%B0/

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() {
 RedisCrossRoomClientStatusCollector.collectCrossRoomStatus(HystrixStatCountTypeEnum.ALL);
 DataComponentCommand<T> majorCommand =
 new DataComponentCommand<T>(MAJOR_READ_COMMAND_KEY, MAJOR_GROUP_KEY, MAJOR_THREAD_POOL_KEY,
 majorTimeOut, majorThreads) {
 protected T run() throws Exception {
 RedisCrossRoomClientStatusCollector.collectCrossRoomStatus(HystrixStatCountTypeEnum.RUN);
 return readMajor();
 }
 public T getBusinessFallback() {
 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) {
 protected T run() throws Exception {
 RedisCrossRoomClientStatusCollector.collectCrossRoomStatus(HystrixStatCountTypeEnum.FALLBACK_RUN);
 return readMinor();
 }
 public T getBusinessFallback() throws RedisCrossRoomReadMinorFallbackException {
 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 {
...
public String get(final String key) {
 return new ReadCommand<String>() {
 protected String readMajor() {
 return majorPipelineCluster.get(key);
 }
 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) {
 protected T run() throws Exception {
 return writeMajor();
 }
 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) {
 protected T run() throws Exception {
 return writeMinor();
 }
 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
 */
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;
 }
 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
 */
public void testRandomReadWithMajorSleep() throws InterruptedException {
 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
public void testRandomWriteWithMajorSleep() throws InterruptedException {
 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;
 }
 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机制来解决这个问题），目前使用打印日志的形式记录。
  • …………..

首页 - Wiki
Copyright © 2011-2025 iteam. Current version is 2.144.1. UTC+08:00, 2025-07-28 09:45
浙ICP备14020137号-1 $访客地图$