2. 程式人生 > >Redis哨兵原理總結（四）：原始碼分析

Redis哨兵原理總結（四）：原始碼分析

阿新 • • 發佈：2019-01-09

目錄

 

本博文主要總結關於哨兵的一些理論知識，主要關注點有一下幾個方面：

一、哨兵解決了什麼問題？

二、哨兵是如何解決“問題一”的？

三、如何使用哨兵？

四、Redis Sentinel客戶端實現的原理是什麼？Java如何操作Redis Sentinel？

四、Redis Sentinel客戶端實現的原理是什麼？Java如何操作Redis Sentinel？

首先，看JedisSentinelPool的建構函式。



Jedis Version：2.9.0

sentinel建構函式：

/**
  *Set<String> sentinels：哨兵集合，格式為IP:PORT
  */
public JedisSentinelPool(String masterName, Set<String> sentinels, GenericObjectPoolConfig poolConfig, int timeout, String password) {
	this(masterName, sentinels, poolConfig, timeout, password, 0);
}
	
	
public JedisSentinelPool(String masterName, Set<String> sentinels, GenericObjectPoolConfig poolConfig, int connectionTimeout, int soTimeout, String password, int database, String clientName) {
	this.connectionTimeout = 2000;
	this.soTimeout = 2000;
	this.database = 0;
	this.masterListeners = new HashSet();
	this.log = Logger.getLogger(this.getClass().getName());
	this.poolConfig = poolConfig;
	this.connectionTimeout = connectionTimeout;
	this.soTimeout = soTimeout;
	this.password = password;
	this.database = database;
	this.clientName = clientName;
	HostAndPort master = this.initSentinels(sentinels, masterName);
	this.initPool(master);
}

 再看，如何初始化哨兵的，initSentinels（sentinels, masterName）。

private HostAndPort initSentinels(Set<String> sentinels, String masterName) {
	HostAndPort master = null;
	boolean sentinelAvailable = false;
	this.log.info("Trying to find master from available Sentinels...");
	Iterator var5 = sentinels.iterator();

	String sentinel;
	HostAndPort hap;
	while(var5.hasNext()) {
		sentinel = (String)var5.next();
		hap = HostAndPort.parseString(sentinel);
		this.log.fine("Connecting to Sentinel " + hap);
		Jedis jedis = null;

		try {
			jedis = new Jedis(hap.getHost(), hap.getPort());
			List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
			sentinelAvailable = true;
			if (masterAddr != null && masterAddr.size() == 2) {
				master = this.toHostAndPort(masterAddr);
				this.log.fine("Found Redis master at " + master);
				break;
			}

			this.log.warning("Can not get master addr, master name: " + masterName + ". Sentinel: " + hap + ".");
		} catch (JedisException var13) {
			this.log.warning("Cannot get master address from sentinel running @ " + hap + ". Reason: " + var13 + ". Trying next one.");
		} finally {
			if (jedis != null) {
				jedis.close();
			}

		}
	}

	if (master == null) {
		if (sentinelAvailable) {
			throw new JedisException("Can connect to sentinel, but " + masterName + " seems to be not monitored...");
		} else {
			throw new JedisConnectionException("All sentinels down, cannot determine where is " + masterName + " master is running...");
		}
	} else {
		this.log.info("Redis master running at " + master + ", starting Sentinel listeners...");
		var5 = sentinels.iterator();

		while(var5.hasNext()) {
			sentinel = (String)var5.next();
			// 獲取哨兵的IP和PORT
			hap = HostAndPort.parseString(sentinel);
			// 建立一個執行緒建立監聽器，對每個哨兵進行監聽
			JedisSentinelPool.MasterListener masterListener = new JedisSentinelPool.MasterListener(masterName, hap.getHost(), hap.getPort());
			// 設定為守護執行緒
			masterListener.setDaemon(true);
			this.masterListeners.add(masterListener);
			masterListener.start();
		}
		return master;
	}
}
 
private HostAndPort initSentinels(Set<String> sentinels, String masterName) {
	HostAndPort master = null;
	boolean sentinelAvailable = false;
	this.log.info("Trying to find master from available Sentinels...");
	Iterator var5 = sentinels.iterator();

	String sentinel;
	HostAndPort hap;
	while(var5.hasNext()) {
		sentinel = (String)var5.next();
		hap = HostAndPort.parseString(sentinel);
		this.log.fine("Connecting to Sentinel " + hap);
		Jedis jedis = null;

		try {
			jedis = new Jedis(hap.getHost(), hap.getPort());
			List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
			sentinelAvailable = true;
			if (masterAddr != null && masterAddr.size() == 2) {
				master = this.toHostAndPort(masterAddr);
				this.log.fine("Found Redis master at " + master);
				break;
			}

			this.log.warning("Can not get master addr, master name: " + masterName + ". Sentinel: " + hap + ".");
		} catch (JedisException var13) {
			this.log.warning("Cannot get master address from sentinel running @ " + hap + ". Reason: " + var13 + ". Trying next one.");
		} finally {
			if (jedis != null) {
				jedis.close();
			}

		}
	}

	if (master == null) {
		if (sentinelAvailable) {
			throw new JedisException("Can connect to sentinel, but " + masterName + " seems to be not monitored...");
		} else {
			throw new JedisConnectionException("All sentinels down, cannot determine where is " + masterName + " master is running...");
		}
	} else {
		this.log.info("Redis master running at " + master + ", starting Sentinel listeners...");
		var5 = sentinels.iterator();

		while(var5.hasNext()) {
			sentinel = (String)var5.next();
			// 獲取哨兵的IP和PORT
			hap = HostAndPort.parseString(sentinel);
			// 建立一個執行緒建立監聽器，對每個哨兵進行監聽
			JedisSentinelPool.MasterListener masterListener = new JedisSentinelPool.MasterListener(masterName, hap.getHost(), hap.getPort());
			// 設定為守護執行緒
			masterListener.setDaemon(true);
			this.masterListeners.add(masterListener);
			masterListener.start();
		}
		return master;
	}
}

對master監聽的監聽器的原始碼，建立一個執行緒監聽master。

protected class MasterListener extends Thread {
	protected String masterName;
	protected String host;
	protected int port;
	protected long subscribeRetryWaitTimeMillis;
	protected volatile Jedis j;
	protected AtomicBoolean running;

	protected MasterListener() {
		this.subscribeRetryWaitTimeMillis = 5000L;
		this.running = new AtomicBoolean(false);
	}

	public MasterListener(String masterName, String host, int port) {
		super(String.format("MasterListener-%s-[%s:%d]", masterName, host, port));
		this.subscribeRetryWaitTimeMillis = 5000L;
		this.running = new AtomicBoolean(false);
		this.masterName = masterName;
		this.host = host;
		this.port = port;
	}

	public MasterListener(String masterName, String host, int port, long subscribeRetryWaitTimeMillis) {
		this(masterName, host, port);
		this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis;
	}

	public void run() {
		this.running.set(true);

		while(this.running.get()) {
			this.j = new Jedis(this.host, this.port);

			try {
				if (!this.running.get()) {
					break;
				}
				// 訂閱+switch-master頻道。
				// 如果哨兵完成了主從切換，會在+switch-master頻道釋出這行資訊：pmaster 172.16.0.149 6379 172.16.0.34 6379
				// 然後對訊息“pmaster 172.16.0.149 6379 172.16.0.34 6379”以“ ”分割得到一個String型別的陣列，陣列的第4個元素就是新的master的IP，第5個元素就是新的master的port，第1個元素就是新的master的名稱。
				this.j.subscribe(new JedisPubSub() {
					public void onMessage(String channel, String message) {
						JedisSentinelPool.this.log.fine("Sentinel " + MasterListener.this.host + ":" + MasterListener.this.port + " published: " + message + ".");
						String[] switchMasterMsg = message.split(" ");
						if (switchMasterMsg.length > 3) {
							if (MasterListener.this.masterName.equals(switchMasterMsg[0])) {
								JedisSentinelPool.this.initPool(JedisSentinelPool.this.toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4])));
							} else {
								JedisSentinelPool.this.log.fine("Ignoring message on +switch-master for master name " + switchMasterMsg[0] + ", our master name is " + MasterListener.this.masterName);
							}
						} else {
							JedisSentinelPool.this.log.severe("Invalid message received on Sentinel " + MasterListener.this.host + ":" + MasterListener.this.port + " on channel +switch-master: " + message);
						}
					}
				}, new String[]{"+switch-master"});
			} catch (JedisConnectionException var8) {
				if (this.running.get()) {
					JedisSentinelPool.this.log.log(Level.SEVERE, "Lost connection to Sentinel at " + this.host + ":" + this.port + ". Sleeping 5000ms and retrying.", var8);

					try {
						Thread.sleep(this.subscribeRetryWaitTimeMillis);
					} catch (InterruptedException var7) {
						JedisSentinelPool.this.log.log(Level.SEVERE, "Sleep interrupted: ", var7);
					}
				} else {
					JedisSentinelPool.this.log.fine("Unsubscribing from Sentinel at " + this.host + ":" + this.port);
				}
			} finally {
				this.j.close();
			}
		}
	}
}

 

Redis訊息訂閱與釋出：

客戶端可以將 Sentinel 看作是一個只提供了訂閱功能的 Redis 伺服器：

你不可以使用 PUBLISH 命令向這個伺服器傳送資訊， 但你可以用 SUBSCRIBE 命令或者 PSUBSCRIBE 命令， 通過訂閱給定的頻道來獲取相應的事件提醒。

一個頻道能夠接收和這個頻道的名字相同的事件。

比如說， 名為 +sdown 的頻道就可以接收所有例項進入主觀下線（SDOWN）狀態的事件。通過執行 PSUBSCRIBE * 命令可以接收所有事件資訊。
    +switch-master <master name> <oldip> <oldport> <newip> <newport> ：配置變更，主伺服器的 IP 和地址已經改變。

這是絕大多數外部使用者都關心的資訊。

可以看出，我們使用Sentinel命令和釋出訂閱兩種機制就能很好的實現和客戶端的整合整合：使用get-master-addr-by-name和slaves指令可以獲取當前的Master和Slaves的地址和資訊；而當發生故障轉移時，即Master發生切換，可以通過訂閱的+switch-master事件獲得最新的Master資訊。

——————————————————Redis哨兵原理總結完結——————————————————

訂閱與釋出的基本原理：
https://redisbook.readthedocs.io/en/latest/feature/pubsub.html

 

相關推薦

Redis哨兵原理總結（四）：原始碼分析

目錄   本博文主要總結關於哨兵的一些理論知識，主要關注點有一下幾個方面： 一、哨兵解決了什麼問題？ 二、哨兵是如何解決“問題一”的？ 三、如何使用哨兵？ 四、Redis Sentinel客戶端實現的原理是什麼？Java如何操作Redis Sentinel？

Redis哨兵原理總結（一）

本博文主要總結關於哨兵的一些理論知識，主要關注點有一下幾個方面： 一、哨兵解決了什麼問題？ 二、哨兵是如何解決“問題一”的？ 三、如何使用哨兵？ 四、Redis Sentinel客戶端實現的原理是什麼？Java如何操作Redis Sentinel？ 一、哨兵解決

Redis哨兵原理總結（三）

目錄   三、如何使用哨兵？ 3.1、哨兵環境部署 3.2、主節點、從節點、哨兵節點配置檔案解釋  3.3、哨兵節點日誌分析 本博文主要總結關於哨兵的一些理論知識，主要關注點有一下幾個方面： 一、哨兵解決了什麼問題？ 二、哨兵是如何解決“問

Redis哨兵原理總結（二）

目錄 二、哨兵是如何解決“問題一”的？ 2.1 、主從模式哨兵部署 2.2、三個定時監控任務  2.3、主觀下線與客觀下線 2.4、領導者哨兵節點選舉 2.5、故障轉移 本博文主要總結關於哨兵的一些理論知識，主要關注點有一下幾個方面： 一、哨兵解決了

Maven學習總結（四）：更改maven的編碼格式方式

fis mtu avd WAD nav npe env unp bcp 安裝系統之後，一般中文系統默認字符集是GBK。我們安裝的軟件一般都繼承使用操作系統的默認字符集。所以當在中文XP或者win7系統開發，在使用maven（mvn compile）編譯項目的時候，就會出現“

JavaWeb學習總結（四）：Servlet開發（二）

一、ServletConfig講解 1.1、配置Servlet初始化引數 　　在Servlet的配置檔案web.xml中，可以使用一個或多個<init-param>標籤為servlet配置一些初始化引數。 例如： 1 <servlet&g

ElasticSearch學習總結（四）：分散式特性

本文主要對Elasticsearch的分散式相關特性進行總結 1. 分片與副本 1.1 分片&副本 索引分片機制用來儲存超過單個節點儲存容量的資料，分片副本用來應對不斷攀升的吞吐量以及確保資料的安全性。 當一個節點的主分片丟失，ElasticSearch可以把任意一個

C#工作總結（四）：迴圈引用的思考

一. 引子（Introduction）         大概在半年前，曾與公司某個同事爭吵一個迴圈引用的問題。當時吵的面紅耳赤，這幾天想把當時，為了吵贏那次問題，做的功課給寫下來。         然後是，以前公司部門經理吹噓某部門大佬在移交客戶程式碼中解決迴圈引用的

微信第三方平臺開發經驗總結（四）：重定向到授權頁

重定向到授權頁 步驟2：引入使用者進入授權頁 第三方平臺方可以在自己的網站:中放置“微信公眾號授權”或者“小程式授權”的入口，引導公眾號和小程式管理員進入授權頁。授權頁網址為https://m

程式設計師的踩坑經驗總結（四）：死鎖

死鎖也是程式設計師最常見的問題之一了，但是死鎖跟記憶體洩露不同，原理和原因都相對簡單，簡單說就是你等我，我也等你，就這麼耗著！ 但死鎖的影響有時比記憶體洩露更嚴重。記憶體洩露主要是漸進式的，可能重啟一下就可以從頭開始了。而死鎖是重啟不了，這只是直接影響而已。死鎖一般會出現某個功能或者操作無反應，可能進一步沒有

Caffe原始碼（四）：math_functions 分析

轉自：https://blog.csdn.net/seven_first/article/details/47378697#1-caffecpugemm-%E5%87%BD%E6%95%B0 主要函式 math_function 定義了caffe 中用到的一些矩陣操作和數值計算的一些函式，這

ESP8266 SDK學習筆記（四）：Makefile分析

主要內容 本章我們來分析SDK中的Makefile，同時學習如何建立自己的原始碼資料夾。 預備知識 一份程式碼工程編譯成二進位制可執行檔案，需要經過四個步驟：預處理、編譯、彙編和連結。 進行編譯需要有編譯器，Linux環境下使用的是gcc，而對應的，

Shiro許可權管理框架（四）：深入分析Shiro中的Session管理

其實關於Shiro的一些學習筆記很早就該寫了，因為懶癌和拖延症晚期一直沒有落實，直到今天公司的一個專案碰到了在叢集環境的單點登入頻繁掉線的問題，為了解決這個問題，Shiro相關的文件和教程沒少翻。最後問題解決了，但我覺得我也是時候來做一波Shiro學習筆記了。 本篇是Shiro系列第四篇，Shiro中的過濾器

併發程式設計（四）—— ThreadLocal原始碼分析及記憶體洩露預防

今天我們一起探討下ThreadLocal的實現原理和原始碼分析。首先，本文先談一下對ThreadLocal的理解，然後根據ThreadLocal類的原始碼分析了其實現原理和使用需要注意的地方，最後給出了兩個應用場景。相信本文一定能讓大家完全瞭解ThreadLocal。 ThreadL

Spring5原始碼分析系列（四）Spring5原始碼分析2

本文緊接上文Spring5原始碼分析1，講解基於XML的依賴注入，文章參考自Tom老師視訊，下一篇文章將介紹基於Annotation的依賴注入。 基於XML的依賴注入 1、依賴注入發生的時間 當SpringIOC容器完成了Bean定義資源的定位、載入和解析註冊以後，IO

DelayQueue系列（一）：原始碼分析

本文原發表於簡書DelayQueue之原始碼分析。 本文將會對DelayQueue做一個簡單的介紹，並提供部分原始碼的分析。 DelayQueue的特性基本上由BlockingQueue、PriorityQueue和Delayed的特性來決定的。 簡而言之，DelayQueue是通過Delayed，使得

Java併發包原始碼學習之AQS框架（四）AbstractQueuedSynchronizer原始碼分析

經過前面幾篇文章的鋪墊，今天我們終於要看看AQS的廬山真面目了，建議第一次看AbstractQueuedSynchronizer 類原始碼的朋友可以先看下我前面幾篇文章： 分析原始碼是非常枯燥乏味的一件事，其實程式碼本身其實就是最好的說明了，因此基本都是貼出一些程式碼加上一些註釋， 因為Abstract

嵌入式Linux——nand flash 驅動（三）：原始碼分析

再次宣告：本文是看過一些文章後寫的，如果與你的文章有相同的地方，敬請告知，如果對你有幫助，是我的榮幸。 接下來的這篇文章我們將要分析一下nand flash在S3C2440中的驅動函式。下面我們以一張圖來引入：         從上圖可以看出，MTD裝置層與原始裝置層

深入理解阻塞佇列（四）——LinkedBlockingDeque原始碼分析

LinkedBlockingDeque是一個基於連結串列的雙端阻塞佇列。和LinkedBlockingQueue類似，區別在於該類實現了Deque介面，而LinkedBlockingQueue實現了Queue介面。該類的繼承關係如下圖： 本文將與Lin

HLS學習（四）HLSDownloader原始碼分析（3）總體流程

總體流程 下載一個媒體檔案的流程 1、初始化 2、根據URL下載m3u8檔案（即PlayList檔案） 3、判斷m3u8檔案的型別，是Master PlayList還是Media PlayList 4、如果是Master PlayList，那麼分析這個PlayList