背景

接手了一個系統，該使用了HAProxy + Sentinel +redis方案，該方案在redis發生主從切換后，因為應用層用的是redis的連接池，老連接仍然是連接的redis的“老主節點”。進行寫操作的時候，會拋出異常。
org.springframework.data.redis.RedisSystemException: Error in execution; nested exception is io.lettuce.core.RedisCommandExecutionException: READONLY You can’t write against a read only replica.

該方案如下

redis使用使用哨兵模式進行組網，哨兵負責主節點的故障轉移。
HAProxy作為Redis集群的代理(HAProxy工作臺TCP層)，屏蔽底層redis的組網細節，對上層應用來看就是單節點的redis。
從下面的配置可以看到，每次應用層創建新連接時，HAproxy會輪訓所有節點，如果探測到節點為主節點，則代應用層向其發起連接。

defaults
  mode tcp

frontend redis
bind *:16382 name redis
  default_backend redis
  
backend redis
option tcp-check
tcp-check connect
tcp-check send AUTH\ xxxxxxxxx\r\n
tcp-check expect string +OK
tcp-check send PING\r\n
tcp-check expect string +PONG
tcp-check send info\ replication\r\n
tcp-check expect string role:master
tcp-check send QUIT\r\n
tcp-check expect string +OK
server redis_6382_1 xx.xx.xx.xx:6382 check inter 1s
server redis_6382_2 xx.xx.xx.xx:6382 check inter 1s
server redis_6382_3 xx.xx.xx.xx:6382 check inter 1s
server redis_6382_4 xx.xx.xx.xx:6382 check inter 1s

該方案的問題

不像http協議連接被設計成無狀態。redis的協議較簡單，且是有狀態的。在執行操作前得先認證（認證是可選的），認證完了后，可以長期保持連接，進行交互，除非客戶端主動斷開連接(不深究，可能服務端設置了tcp參數，長期無響應的連接會被服務端關閉)。
所以jedis設計了連接池，創建出來的連接只要能夠正常訪問，連接每次用完后會被回收到連接池中進行復用。
哨兵模式通過心跳檢測redis主節點是否發生故障，然后進行故障轉移（主從切換）。發生主從切換時，并不一定是redis掛了，有時候可能redis的負債過重，無法及時響應哨兵的PING命令。
這個時候雖然哨兵認為redis掛了，但是應用層的之前建立的連接還是能夠正常訪問的，所以老的連接不會被銷毀。但發生主從切換后，使用老連接進行寫操作時就會導致異常，因為此時老連接連的是“老主節點”，發生主從切換后，“老主節點”變成從節點且只讀。

解釋：應用層的之前建立的連接還是能夠正常訪問
第一，應用層查詢頻率不一定有哨兵的心跳檢測頻率高，所以redis負載過重的時候，部分老連接并沒有發起訪問。
第二，應用層查詢的超時時間設置的比哨兵的心跳檢測超時時間長。即使響應很慢，應用層仍然認為連接是健康的。

問題復現

人工模擬哨兵的主從切換。

1. 去掉Sentinel的監控（kill掉Sentinel進程）
2. 找一個從節點，執行slaveof no one
3. 剩余的節點執行，slaveof 新節點

方案選型

Jedis目前支持哨兵模式了，不過我們系統無法容忍長時間的主備變更信息延時通知到應用層，因此需要審視jedis的源碼，確認集群發生主從切換后，是否能夠快速通知到應用層并清理老連接。
測試代碼如下：

	public static void main(String[] args) throws JsonProcessingException {
		// SpringApplication.run(DemoApplication.class, args);
		RedisTemplate<String,String> redisTemplate = new RedisTemplate();
		Set<String> setRedisNode = new HashSet<>();
		# 哨兵的IP:Port
		setRedisNode.add("xx.xx.xx.xx:26379");
		setRedisNode.add("xx.xx.xx.xx:26381");
		setRedisNode.add("xx.xx.xx.xx:26382");
		setRedisNode.add("xx.xx.xx.xx:26383");
		# "mymaster" 對應sentinel.conf中"sentinel monitor mymaster xx.xx.xx.xx 6383 2"
		RedisSentinelConfiguration redisSentinelConfiguration = new RedisSentinelConfiguration("mymaster", setRedisNode);
		JedisPoolConfig config = new JedisPoolConfig();
		JedisConnectionFactory connectionFactory = new JedisConnectionFactory(redisSentinelConfiguration,config);
		connectionFactory.afterPropertiesSet();
		redisTemplate.setConnectionFactory(connectionFactory);
		redisTemplate.afterPropertiesSet();
		redisTemplate.opsForValue().set("aaa","bbb");
		while(true){}
	}

跟蹤調用鏈，發現JedisConnectionFactory (afterPropertiesSet方法)在初始化的過程中會做以下動作：

1. 向哨兵進程查詢當前主節點
2. 使用異步線程監聽哨兵發過來的事件，如果是主從切換事件，則立馬更新主節點，并清理連接池。

 # 代碼有刪減
  public JedisSentinelPool(String masterName, Set<HostAndPort> sentinels,
      final GenericObjectPoolConfig<Jedis> poolConfig, final JedisFactory factory,
      final JedisClientConfig sentinelClientConfig) {
    # 查詢當前主節點
    HostAndPort master = initSentinels(sentinels, masterName);
    # 設置當前主節點
    initMaster(master);
  }

JedisSentinelPool#initSentinels方法詳解，代碼有刪減

  private HostAndPort initSentinels(Set<HostAndPort> sentinels, final String masterName) {
    HostAndPort master = null;
    boolean sentinelAvailable = false;
    
    # 防止部分哨兵不可用
    for (HostAndPort sentinel : sentinels) {
      Jedis jedis = new Jedis(sentinel, sentinelClientConfig)) 
      # 連接哨兵進程并通過"SENTINEL get-master-addr-by-name mymaster"命令查詢當前主節點
      List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
      sentinelAvailable = true;
      if (masterAddr == null || masterAddr.size() != 2) {
          continue;
      }
      master = toHostAndPort(masterAddr);
      break;
    }
    # MasterListener是Thread的子類
    # 是通過異步線程去檢測主從切換事件，然后及時更新主節點，清理連接池
    for (HostAndPort sentinel : sentinels) {
      MasterListener masterListener = new MasterListener(masterName, sentinel.getHost(), sentinel.getPort());
      masterListener.setDaemon(true);
      masterListeners.add(masterListener);
      masterListener.start();
    }
    return master;
  }

MasterListener 的run方法，代碼有刪減

MasterListener是JedisSentinelPool內部類，因此能夠操作JedisSentinelPool的變量(masterName)和方法(initMaster)
感興趣的同學可以繼續深入JedisPubSub代碼，可以發現如下：

• 主從切換事件是哨兵進程主動推送過來的，所以能夠保證實時性
  因為redis協議是雙工的，所以服務端可以主動推數據給客戶端。
  監聽事件的過程就是，創建socket連接，然后讀取數據進行解析。
  不發生主從切換事件時，沒有數據推送過來，線程會阻塞在read操作上面。所以雖然run方法是死循環，但是并不會占用cpu時間。

    public void run() {
      while (running.get()) {
          final HostAndPort hostPort = new HostAndPort(host, port);
          # 連接本MasterListener關注的哨兵進程
          j = new Jedis(hostPort, sentinelClientConfig);
          
          # 監聽+switch-master事件
          j.subscribe(new JedisPubSub() {
            @Override
            public void onMessage(String channel, String message) {
              String[] switchMasterMsg = message.split(" ");
              if (masterName.equals(switchMasterMsg[0])) {
                initMaster(toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4]))); 
              } 
            }
          }, "+switch-master");

  }

JedisSentinelPool#initMaster

如果和當前主節點信息不一致，這更新主節點信息，并清理連接池

  private void initMaster(HostAndPort master) {
    synchronized (initPoolLock) {
      if (!master.equals(currentHostMaster)) {
        currentHostMaster = master;
        factory.setHostAndPort(currentHostMaster);
        clearInternalPool();
      }
    }
  }

jedis的哨兵模式并沒有實現讀寫分離，僅僅只與主節點建立連接。所以沒辦法完全挖掘集群的性能。

可以通過redisTemplate.opsForValue().set(“aaa”,“bbb”);去跟蹤新連接的創建過程（非復用連接池中的連接）。
最終會追蹤到JedisConnectionFactory#fetchJedisConnector方法

	protected Jedis fetchJedisConnector() {
	# jedis會開啟連接池
	    if (getUsePool() && pool != null) {
			return pool.getResource();
		}
		Jedis jedis = createJedis();
		jedis.connect();
		return jedis;
	}

JedisSentinelPool負責連接的創建，JedisSentinelPool持有主節點信息

  public Jedis getResource() {
    while (true) {
      Jedis jedis = super.getResource();
      jedis.setDataSource(this);
      final HostAndPort master = currentHostMaster;
      final HostAndPort connection = new HostAndPort(jedis.getClient().getHost(), jedis.getClient()
          .getPort());

      if (master.equals(connection)) {
        return jedis;
      } else {
        returnBrokenResource(jedis);
      }
    }
  }