master選舉
1、使用場景及結構
現在很多時候我們的服務需要7*24小時工作，假如一台機器掛了，我們希望能有其它機器頂替它繼續工作。此類問題現在多采用master-salve模式，也就是常說的主從模式，正常情況下主機提供服務，備機負責監聽主機狀態，當主機異常時，可以自動切換到備機繼續提供服務(這里有點兒類似於數據庫主庫跟備庫，備機正常情況下只監聽，不工作)，這個切換過程中選出下一個主機的過程就是master選舉。
對於以上提到的場景，傳統的解決方式是采用一個備用節點，這個備用節點定期給當前主節點發送ping包，主節點收到ping包后會向備用節點發送應答ack，當備用節點收到應答，就認為主節點還活着，讓它繼續提供服務，否則就認為主節點掛掉了，自己將開始行使主節點職責。如圖1所示：
圖1
但這種方式會存在一個隱患，就是網絡故障問題。看一下圖2：
圖2
也就是說，我們的主節點並沒有掛掉，只是在備用節點ping主節點，請求應答的時候發生網絡故障，這樣我們的備用節點同樣收不到應答，就會認為主節點掛掉，然后備機會啟動自己的master實例。這樣就會導致系統中有兩個主節點，也就是雙master。出現雙master以后，我們的從節點會將它做的事情一部分匯報給主節點，一部分匯報給備用節點，這樣服務就亂套了。為了防止這種情況出現，我們可以考慮采用zookeeper，雖然它不能阻止網絡故障的出現，但它能保證同一時刻系統中只存在一個主節點。我們來看zookeeper是怎么實現的：
在此處，搶主程序是包含在服務程序中，需要程序員來手動寫搶主邏輯的。
一點額外的話：zookeeper自己在集群環境下的搶主算法有三種，可以通過配置文件來設定，默認采用FastLeaderElection，不作贅述；此處主要討論集群環境中，應用程序利用master的特點，自己選主的過程。程序自己選主，每個人都有自己的一套算法，有采用“最小編號”的，有采用類似“多數投票”的，各有優劣，本文的算法僅作演示理解使用：
結構圖：
結構圖解釋：左側樹狀結構為zookeeper集群，右側為程序服務器。所有的服務器在啟動的時候，都會訂閱zookeeper中master節點的刪除事件，以便在主服務器掛掉的時候進行搶主操作；所有服務器同時會在servers節點下注冊一個臨時節點（保存自己的基本信息），以便於應用程序讀取當前可用的服務器列表。
選主原理介紹：zookeeper的節點有兩種類型，持久節點跟臨時節點。臨時節點有個特性，就是如果注冊這個節點的機器失去連接(通常是宕機)，那么這個節點會被zookeeper刪除。選主過程就是利用這個特性，在服務器啟動的時候，去zookeeper特定的一個目錄下注冊一個臨時節點(這個節點作為master，誰注冊了這個節點誰就是master)，注冊的時候，如果發現該節點已經存在，則說明已經有別的服務器注冊了(也就是有別的服務器已經搶主成功)，那么當前服務器只能放棄搶主，作為從機存在。同時，搶主失敗的當前服務器需要訂閱該臨時節點的刪除事件，以便該節點刪除時(也就是注冊該節點的服務器宕機了或者網絡斷了之類的)進行再次搶主操作。從機具體需要去哪里注冊服務器列表的臨時節點，節點保存什么信息，根據具體的業務不同自行約定。選主的過程，其實就是簡單的爭搶在zookeeper注冊臨時節點的操作，誰注冊了約定的臨時節點，誰就是master。
ps：本文的例子中，並未用到結構圖server節點下的數據。但換一種算法或者業務場景就會用到，算法比如提到的最小編號，主要邏輯是主節點掛掉后，從節點里邊編號最小的成為主節點，此時會用到該節點內容。換一種業務場景：集群環境中，有很多任務要處理， 主節點負責接收任務，並根據一定算法將任務分配到不同的機器上執行；這種情況下，主節點跟從節點的職責也是不同的，主節點掛掉也會涉及到從節點進行master選舉的問題。這種情況下，很顯然，作為主節點需要知道當前有多少個從節點還活着，那么此時也會需要用到servers節點下的數據了。
架構圖：
1)、左邊區域代表zk集群
2)、右邊代表3台工作服務器
它們在各自啟動過程中首先會去zk集群的Servers節點下創建臨時節點，並把自己的基本信息寫入到臨時節點，這個過程叫做服務注冊。系統中的其他服務可以通過獲取Servers節點的子節點列表來了解當前系統哪些服務器可用。這個過程叫做服務發現。
接着這些服務器會去嘗試創建Master節點，誰能創建成功，誰就作為master向外提供服務。其他機器作為slave，所有的slave必須關注master節點的刪除事件。
一個臨時節點在創建它的會話失效以后會自動的被zk刪除掉，而創建會話的機器宕機會直接導致會話的失效。我們可以監聽master節點的失效來了解master節點是否宕機，一旦宕機，就必須發起新一輪的master選舉。新選舉出的master繼續提供服務。
程序主體流程：
work server在啟動的時候首先會注冊監聽master節點的刪除事件 ，緊接着會嘗試創建master節點，如果可以創建成功，說明自己就是master，如果不能說明當前系統中master節點已存在，其他機器爭搶到了master權利，這個時候我們可以讀取master節點的數據內容，如果可以讀取成功，就把master的基本信息放入自己的內存變量中。如果不能，說明在讀取master的瞬間master宕機了。這個時候需要發起新一輪的master 選舉來爭搶master權利。
應對網絡抖動流程：
系統的核心類：
Work Server：對應架構圖的Work Server，是主工作類。
Running Data：用於描述Work Server的基本信息。
LeaderSelectorZkClient：作為調度器來啟動停止Work Server。
2、編碼實現
主要有兩個類，WorkServer為主服務類，RunningData用於記錄運行數據。因為是簡單的demo，我們只做搶master節點的編碼，對於從節點應該去哪里注冊服務列表信息，不作編碼。
采用zkClient實現，代碼如下：
WorkServer類：
import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import org.I0Itec.zkclient.IZkDataListener; import org.I0Itec.zkclient.ZkClient; import org.I0Itec.zkclient.exception.ZkException; import org.I0Itec.zkclient.exception.ZkInterruptedException; import org.I0Itec.zkclient.exception.ZkNoNodeException; import org.I0Itec.zkclient.exception.ZkNodeExistsException; import org.apache.zookeeper.CreateMode; public class WorkServer { private volatile boolean running = false; //記錄服務器運行狀態 private ZkClient zkClient; //開源客戶端-zk客戶端 private static final String MASTER_PATH = "/master"; //master節點對應在zookeeper中的節點路徑 private IZkDataListener dataListener; //監聽zookeeper中master節點的刪除事件 private RunningData serverData; //集群中當前服務器節點的基本信息 private RunningData masterData; //集群中master節點的基本信息 private ScheduledExecutorService delayExector = Executors.newScheduledThreadPool(1); private int delayTime = 5; public WorkServer(RunningData rd) { this.serverData = rd; this.dataListener = new IZkDataListener() { public void handleDataDeleted(String dataPath) throws Exception {//節點刪除事件 //takeMaster(); if (masterData != null && masterData.getName().equals(serverData.getName())) { takeMaster(); } else { delayExector.schedule(new Runnable() { public void run() { takeMaster(); } }, delayTime, TimeUnit.SECONDS); } } public void handleDataChange(String dataPath, Object data) throws Exception {//節點內容變化事件 } }; } public ZkClient getZkClient() { return zkClient; } public void setZkClient(ZkClient zkClient) { this.zkClient = zkClient; } /** * 服務start方法 * * @throws Exception */ public void start() throws Exception { if (running) { throw new Exception("server has startup..."); } running = true; zkClient.subscribeDataChanges(MASTER_PATH, dataListener); takeMaster(); } /** * 服務stop方法 * * @throws Exception */ public void stop() throws Exception { if (!running) { throw new Exception("server has stoped"); } running = false; delayExector.shutdown(); zkClient.unsubscribeDataChanges(MASTER_PATH, dataListener); releaseMaster(); } /** * 爭搶master權利 */ private void takeMaster() { if (!running) { return; } try { zkClient.create(MASTER_PATH, serverData, CreateMode.EPHEMERAL); masterData = serverData; System.out.println(serverData.getName() + " is master"); // 以下代碼作為演示,每隔5秒鍾釋放master權利 delayExector.schedule(new Runnable() { public void run() { if (checkMaster()) { releaseMaster(); } } }, 5, TimeUnit.SECONDS); } catch (ZkNodeExistsException e) { RunningData runningData = zkClient.readData(MASTER_PATH, true); if (runningData == null) { takeMaster(); } else { masterData = runningData; } } catch (Exception e) { // ignore; } } /** * 釋放master權利 */ private void releaseMaster() { if (checkMaster()) { zkClient.delete(MASTER_PATH); } } /** * 檢測是否是master */ private boolean checkMaster() { try { RunningData eventData = zkClient.readData(MASTER_PATH); masterData = eventData; if (masterData.getName().equals(serverData.getName())) { return true; } return false; } catch (ZkNoNodeException e) { return false; } catch (ZkInterruptedException e) { return checkMaster(); } catch (ZkException e) { return false; } } }
RunningData類：
import java.io.Serializable; public class RunningData implements Serializable { private static final long serialVersionUID = 4260577459043203630L; private Long cid; private String name; public Long getCid() { return cid; } public void setCid(Long cid) { this.cid = cid; } public String getName() { return name; } public void setName(String name) { this.name = name; } }
說明：在實際生產環境中，可能會由於插拔網線等導致網絡短時的不穩定，也就是網絡抖動。由於正式生產環境中可能server在zk上注冊的信息是比較多的，而且server的數量也是比較多的，那么每一次切換主機，每台server要同步的數據量(比如要獲取誰是master，當前有哪些salve等信息，具體視業務不同而定)也是比較大的。那么我們希望，這種短時間的網絡抖動最好不要影響我們的系統穩定，也就是最好選出來的master還是原來的機器，那么就可以避免發現master更換后，各個salve因為要同步數據等導致的zk數據網絡風暴。所以在WorkServer中，54-63行，我們搶主的時候，如果之前主機是本機，則立即搶主，否則延遲5s搶主。這樣就給原來主機預留出一定時間讓其在新一輪選主中占據優勢，從而利於環境穩定。
測試代碼：
import com.sql.zookeeper.common.ZookeeperConstant; import org.I0Itec.zkclient.ZkClient; import org.I0Itec.zkclient.serialize.SerializableSerializer; import java.io.BufferedReader; import java.io.InputStreamReader; import java.util.ArrayList; import java.util.List; public class LeaderSelectorZkClient { //啟動的服務個數 private static final int CLIENT_QTY = 10; public static void main(String[] args) throws Exception { //保存所有zkClient的列表 List<ZkClient> clients = new ArrayList<ZkClient>(); //保存所有服務的列表 List<WorkServer> workServers = new ArrayList<WorkServer>(); try { for (int i = 0; i < CLIENT_QTY; ++i) { //創建zkClient ZkClient client = new ZkClient(ZookeeperConstant.ZK_CONNECTION_STRING, 5000, 5000, new SerializableSerializer()); clients.add(client); //創建serverData RunningData runningData = new RunningData(); runningData.setCid(Long.valueOf(i)); runningData.setName("Client #" + i); //創建服務 WorkServer workServer = new WorkServer(runningData); workServer.setZkClient(client); workServers.add(workServer); workServer.start(); } System.out.println("敲回車鍵退出！\n"); new BufferedReader(new InputStreamReader(System.in)).readLine(); } finally { System.out.println("Shutting down..."); for (WorkServer workServer : workServers) { try { workServer.stop(); } catch (Exception e) { e.printStackTrace(); } } for (ZkClient client : clients) { try { client.close(); } catch (Exception e) { e.printStackTrace(); } } } } }
兩次測試，本地模擬10台server，分別不啟用防止網絡抖動跟啟動防抖動兩次測試結果如下：
未啟動防抖動：
啟用防抖動：
可以看到，未啟用的時候，斷線后重新選出的主機是隨機的，沒規律；啟用防抖動后，每次選出的master都是id為0的機器。
-----------------------------------------------------------------------------------------------------------------------------
至此，我們已經通過編碼實現了簡單的master選舉。但是，不知你有沒有發現，，，，這個選主過程的代碼還真是麻煩啊！
我們只是做一個demo，其中並未考慮復雜的業務場景，但其中的 監聽，異常 等代碼的處理還是讓我覺得有些頭大，怎么辦？Curator應運而生！
為了熟悉Apache Curator，接下來，將用curator來實現master選舉的demo。