分散式協調服務zookeeper總結

1.zookeeper簡介

1.1簡介

Zookeeper是一個分散式協調服務，換言之，就是為用戶的分散式應用程序提供協調服務

  • zookeeper是為別的分散式程序服務的

  • Zookeeper本身就是一個分散式程序（只要有半數以上節點存活，zk就能正常服務）

  • Zookeeper所提供的服務涵蓋：主從協調、伺服器節點動態上下線、統一配置管理、分散式共享鎖、統一名稱服務

  • 雖然說可以提供各種服務，但是zookeeper在底層其實只提供了兩個功能（管理數據和監聽數據）： 管理(存儲，讀取)用戶程序提交的數據； 並為用戶程序提供數據節點監聽服務；

1.2 Zookeeper集群的角色： Leader 和 follower

Advertisements

Zookeeper在配置文件中並沒有指定master和slave，啟動之後通過內部的選舉機制選舉出leader和follower，而且只有一個leader，其他則為follower。zookeeper集群中只要有半數以上節點存活，集群就能提供服務。2.zookeeper集群機制半數機制：集群中半數以上機器存活，集群可用。zookeeper適合裝在奇數台機器上！！！

2.zookeeper安裝與配置

2.1zookeeper安裝

  • 安裝到3台虛擬機上（需要提前安裝好JDK） 將zookeeper壓縮包上傳至/apps/package目錄並解壓

tar -zxvf zookeeper-3.4.5.tar.gz

  • 重命名

    Advertisements 

mv zookeeper-3.4.5 zookeeper（重命名文件夾zookeeper-3.4.5為zookeeper）

  • 修改環境變數

vi /etc/profile添加內容：export ZOOKEEPER_HOME=/apps/package/zookeeperexport PATH=$PATH:$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin

  • 重新編譯文件： source /etc/profile 三台機器都需要修改

2.2 修改配置文件

  • 先複製一份 cd zookeeper/conf cp zoo_sample.cfg zoo.cfg

  • 編輯 vi zoo.cfg

添加內容：dataDir=/apps/package/zookeeper/datadataLogDir=/apps/package/zookeeper/logserver.1=mini1:2888:3888server.2=mini2:2888:3888server.3=mini3:2888:3888

  • 創建文件夾：

cd /apps/package/zookeepermkdir -m 755 datamkdir -m 755 log

  • 在data文件夾下新建myid文件，myid的文件內容為：

cd datavi myid添加內容：1

mini2和mini3伺服器的請修改成2,3，將來會按這個myid選中出leader和follow。

  • 將集群複製到其他機器上

scp -r /apps/package/zookeeper root@mini2:/apps/package/scp -r /apps/package/zookeeper root@mini3:/apps/package/

如果在mini1中ping不通mini2和mini3,需要在hosts文件中配置mini2和mini3的ip地址

  • 修改其他機器的配置文件 到mini2上：修改myid為：2 到mini3上：修改myid為：3 而且/etc/profile的路徑也不要忘了修改

  • 啟動（每台機器）

zkServer.sh startzkServer.sh start-foreground(可以看到啟動日誌)

  • 查看集群狀態

jps（查看進程）zkServer.sh status（查看集群狀態，主從信息）

如果報埠佔用，參考下面鏈接解決：http://blog.csdn.net/u014686180/article/details/51767863

3.zookeeper數據結構和常用操作

3.1zookeeper特性

  • Zookeeper：一個leader，多個follower組成的集群

  • 全局數據一致：每個server保存一份相同的數據副本，client無論連接到哪個server，數據都是一致的

  • 分散式讀寫，更新請求轉發，由leader實施

  • 更新請求順序進行，來自同一個client的更新請求按其發送順序依次執行

  • 數據更新原子性，一次數據更新要麼成功，要麼失敗

  • 實時性，在一定時間範圍內，client能讀到最新數據

3.2zookeeper數據結構

  • 層次化的目錄結構;

  • 每個節點在zookeeper中叫做znode,並且其有一個唯一的路徑標識;

  • 節點Znode可以包含數據和子節點（但是EPHEMERAL類型的節點不能有子節點）;

  • 客戶端應用可以在節點上設置監視器。

3.3節點類型

  • Znode有兩種類型： 短暫（ephemeral）（斷開連接自己刪除） 持久（persistent）（斷開連接不刪除）

  • Znode有四種形式的目錄節點（默認是persistent ） PERSISTENT PERSISTENTSEQUENTIAL（持久序列/test0000000019 ） EPHEMERAL EPHEMERALSEQUENTIAL

  • 創建znode時設置順序標識，znode名稱後會附加一個值，順序號是一個單調遞增的計數器，由父節點維護

  • 在分散式系統中，順序號可以被用於為所有的事件進行全局排序，這樣客戶端可以通過順序號推斷事件的順序

使用客戶端操作節點

  • 使用命令連接zookeeper服務端：

zkCli.sh -主機名(ip):2181如：zkCli.sh -mini2:2181

  • 使用 ls 命令來查看當前 ZooKeeper 中所包含的內容：

 ls /

  • 創建一個新的 znode ，使用 create /zk myData 。這個命令創建了一個新的 znode 節點「 zk 」以及與它關聯的字元串：

create /zk "myData「

  • 我們運行 get 命令來確認 znode 是否包含我們所創建的字元串：

get /zk

-監聽這個節點的變化,當另外一個客戶端改變/zk時,輸出監聽到的變化

get /zk watch

  • 使用set 命令來對 zk 所關聯的字元串進行設置：

set /zk "zsl「

  • 使用delete刪除 znode 節點：

delete /zk

  • 刪除節點(與上面的區別是這個可以刪除目錄)：rmr

rmr /zk

參考文檔：http://www.cnblogs.com/likehua/tag/zookeeper/

4.使用java操作zookeeper的api

  • 首先需要引入zookeeper的jar包，這個jar包需要依賴其它的jar，可以直接到maven倉庫下載。

<dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.10</version> <type>pom</type></dependency>

  • 常用的增刪查改api操作如下： create 在目錄樹中創建一個節點 delete 刪除一個節點 exists 測試是否存在目標節點 get/set data 從目標節點上讀取 / 更新數據 get/set ACL 獲取 / 設置目標節點訪問控制列表信息 get children 檢索一個子節點上的列表 sync 等待要被傳送的數據 使用java操作代碼如下：

public class SimpleZkClient { private static final String CONNECT_URL = "mini1:2181,mini2:2181,mini3:2181"; private static final int SESSION_TIME_OUT = 2000; ZooKeeper zkCli = null; @Before public void init() throws Exception{ zkCli = new ZooKeeper(CONNECT_URL, SESSION_TIME_OUT, new Watcher() { @Override public void process(WatchedEvent event) { System.out.println(event.getType()+"-----------"+event.getPath()); try{ zkCli.getChildren("/", true); }catch (Exception e){ } } }); } /** * @Description 添加節點數據 * @Author 劉俊重 */ @Test public void create() throws Exception{ // 參數1：要創建的節點的路徑 參數2：節點大數據 參數3：節點的許可權 參數4：節點的類型。上傳的數據可以是任何類型，但都要轉成byte[] String s = zkCli.create("/zk", "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } /** * @Description 判斷節點是否存在 * @Author 劉俊重 */ @Test public void isExist() throws Exception{ Stat exists = zkCli.exists("/zk", false); System.out.println(null==exists?"不存在":"存在"); } /** * @Description 獲取節點數據 * @Author 劉俊重 */ @Test public void getData() throws Exception{ byte[] data = zkCli.getData("/zk", false, null); System.out.println("節點數據："+new String(data)); } /** * @Description 遍歷節點數據 * @Author 劉俊重 */ @Test public void getChildren() throws Exception{ List<String> children = zkCli.getChildren("/", false); for(String s : children){ System.out.println("節點名稱："+s); } Thread.sleep(Long.MAX_VALUE); } /** * @Description 刪除節點數據 * @Author 劉俊重 */ @Test public void delete() throws Exception{ //參數2：指定要刪除的版本，-1表示刪除所有版本 zkCli.delete("/zk",-1); this.isExist(); } /** * @Description 更新節點數據 * @Author 劉俊重 */ @Test public void update() throws Exception{ Stat stat = zkCli.setData("/zk", "newtest".getBytes(), -1); this.getData(); }}

Thread.sleep(Long.MAX_VALUE);是為了不讓程序執行完之後立馬結束，讓它睡一會，測試監聽是否實現，同時在process回調函數中寫了收到通知的操作， zkCli.getChildren("/", true);這時如果我們通過linux命令行操作了zookeeper操作節點就會觸發這裡的監聽事件。

5.zookeeper的使用場景

5.1場景一：客戶端動態感知服務端節點變化，實現高可用

現在假設有這樣一種需求：服務端節點有多個，可以動態的上下線；需要讓任意一台客戶端都能實時感知服務端節點的變化，進而連接目前可提供服務的節點。實現思路：我們可以藉助於zookeeper這個第三方中間件，在每台伺服器啟動時都向zookeeper註冊伺服器的節點信息（比如：/servers/server01;/servers/server02）；客戶端每次調用之前都通過getChildren方法獲取最新的伺服器節點信息，同時客戶端在zookeeper註冊監聽，監聽伺服器節點的變化；如果某刻伺服器server01下線了，zookeeper就會發出節點變化通知客戶端，回調process方法拉取最新的伺服器節點信息。服務端代碼如下：

public class DistributeServer { private static final String CONNECT_URL = "mini1:2181,mini2:2181,mini3:2181"; private static final int SESSION_TIME_OUT = 2000; private static final String PARENT_NODE = "/servers"; private ZooKeeper zkCli = null; /** * @Description 獲取連接 * @Author 劉俊重 * @Date 2017/12/13 */ public void getConnect() throws Exception{ zkCli = new ZooKeeper(CONNECT_URL, SESSION_TIME_OUT, new Watcher() { @Override public void process(WatchedEvent event) { System.out.println(event.getType()+"-----------"+event.getPath()); try{ zkCli.getChildren("/", true); }catch (Exception e){ } } }); } /** * @Description 伺服器啟動時向zookeeper註冊服務信息 * @Author 劉俊重 * @Date 2017/12/13 */ public void registerServer(String hostName) throws Exception { String s = zkCli.create(PARENT_NODE + "/", hostName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println("伺服器："+hostName+"已經註冊完畢"); } /** * @Description 模擬實際的業務操作 * @Author 劉俊重 * @Date 2017/12/13 */ public void handelBusiness(String hostname) throws Exception { System.out.println("伺服器："+hostname+"正在處理業務。。。"); Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { DistributeServer server = new DistributeServer(); server.getConnect(); server.registerServer(args[0]); server.handelBusiness(args[0]); }}

客戶端代碼如下：

/** * @author 劉俊重 * @Description 模擬客戶端，拉取最新伺服器節點列表並向zookeeper設置監聽 */public class DistributeClient { private static final String CONNECT_URL = "mini1:2181,mini2:2181,mini3:2181"; private static final int SESSION_TIME_OUT = 2000; private static final String PARENT_NODE = "/servers"; private ZooKeeper zkCli = null; private volatile List<String> serverList = null; /** * @Description 獲取連接 * @Author 劉俊重 * @Date 2017/12/13 */ public void getConnect() throws Exception{ zkCli = new ZooKeeper(CONNECT_URL, SESSION_TIME_OUT, new Watcher() { @Override public void process(WatchedEvent event) { // 收到事件通知后的回調函數（應該是我們自己的事件處理邏輯） System.out.println(event.getType()+"-----------"+event.getPath()); try{ //重新更新伺服器列表，並且註冊了監聽 getServerList(); }catch (Exception e){ } } }); } /** * @Description 獲取伺服器子節點信息，並對父節點進行監聽 * @Author 劉俊重 */ public void getServerList() throws Exception { List<String> children = zkCli.getChildren(PARENT_NODE, true); List<String> servers = new ArrayList<String>(); for(String child : children){ // child只是子節點的節點名 byte[] data = zkCli.getData(PARENT_NODE + "/" + child, false, null); servers.add(new String(data)); } //把servers賦值給成員變數serverList，以提供給各業務線程使用 serverList = servers; System.out.println("節點數據："+serverList); } /** * @Description 模擬實際的業務操作 * @Author 劉俊重 * @Date 2017/12/13 */ public void handelBusiness() throws Exception { System.out.println("客戶端開始工作。。。"); Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { DistributeClient client = new DistributeClient(); client.getConnect(); client.getServerList(); client.handelBusiness(); }}

5.2場景二：分散式鎖實現

假設現在集群中有50台機器對某台機器上的同一文件進行修改，如何才能保證這個文件不會被寫亂呢，使用java中的synchronized鎖肯定是不行的，因為這個鎖是對某個程序而言的，而我們這根本就不是在一個伺服器上，怎麼會鎖的住，用zookeeper實現的分散式鎖可以實現。設計思路：伺服器啟動時都去zookeeper上註冊一個「短暫+序號」的znode節點（如/lock/1;/lock/2），並設置監聽父節點變化；獲取到父節點下所有子節點，並比較序號的大小；約定比如序號最小的獲取鎖，去操作某一文件，操作完成後刪除自己的節點（相當於釋放鎖），並註冊一個新的「短暫+序號」的znode節點；其它程序收到zookeeper發送的節點變化的通知之後，去比較序號的大小，看誰獲得新鎖。

public class DistributedClientLock { // 會話超時 private static final int SESSION_TIMEOUT = 2000; // zookeeper集群地址 private String hosts = "mini1:2181,mini2:2181,mini3:2181"; private String groupNode = "locks"; private String subNode = "sub"; private boolean haveLock = false; private ZooKeeper zk; // 記錄自己創建的子節點路徑 private volatile String thisPath; /** * 連接zookeeper */ public void connectZookeeper() throws Exception { zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() { @Override public void process(WatchedEvent event) { try { // 判斷事件類型，此處只處理子節點變化事件 if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) { //獲取子節點，並對父節點進行監聽 List<String> childrenNodes = zk.getChildren("/" + groupNode, true); String thisNode = thisPath.substring(("/" + groupNode + "/").length()); // 去比較是否自己是最小id Collections.sort(childrenNodes); if (childrenNodes.indexOf(thisNode) == 0) { //訪問共享資源處理業務，並且在處理完成之後刪除鎖 doSomething(); //重新註冊一把新的鎖 thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); } } } catch (Exception e) { e.printStackTrace(); } } }); // 1、程序一進來就先註冊一把鎖到zk上 thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); // wait一小會，便於觀察 Thread.sleep(new Random().nextInt(1000)); // 從zk的鎖父目錄下，獲取所有子節點，並且註冊對父節點的監聽 List<String> childrenNodes = zk.getChildren("/" + groupNode, true); //如果爭搶資源的程序就只有自己，則可以直接去訪問共享資源 if (childrenNodes.size() == 1) { doSomething(); thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); } } /** * 處理業務邏輯，並且在最後釋放鎖 */ private void doSomething() throws Exception { try { System.out.println("gain lock: " + thisPath); Thread.sleep(2000); } finally { System.out.println("finished: " + thisPath); //釋放鎖 zk.delete(this.thisPath, -1); } } public static void main(String[] args) throws Exception { DistributedClientLock dl = new DistributedClientLock(); dl.connectZookeeper(); Thread.sleep(Long.MAX_VALUE); }}

參考文檔：http://www.cnblogs.com/likehua/tag/zookeeper/

6 zookeeper的選舉機制

6.1全新集群paxos

以一個簡單的例子來說明整個選舉的過程.假設有五台伺服器組成的zookeeper集群,它們的id從1-5,同時它們都是最新啟動的,也就是沒有歷史數據,在存放數據量這一點上,都是一樣的.假設這些伺服器依序啟動,來看看會發生什麼.1) 伺服器1啟動,此時只有它一台伺服器啟動了,它發出去的報沒有任何響應,所以它的選舉狀態一直是LOOKING狀態2) 伺服器2啟動,它與最開始啟動的伺服器1進行通信,互相交換自己的選舉結果,由於兩者都沒有歷史數據,所以id值較大的伺服器2勝出,但是由於沒有達到超過半數以上的伺服器都同意選舉它(這個例子中的半數以上是3),所以伺服器1,2還是繼續保持LOOKING狀態.3) 伺服器3啟動,根據前面的理論分析,伺服器3成為伺服器1,2,3中的老大,而與上面不同的是,此時有三台伺服器選舉了它,所以它成為了這次選舉的leader.4) 伺服器4啟動,根據前面的分析,理論上伺服器4應該是伺服器1,2,3,4中最大的,但是由於前面已經有半數以上的伺服器選舉了伺服器3,所以它只能接收當小弟的命了.5) 伺服器5啟動,同4一樣,當小弟.

6.2非全新集群的選舉機制(數據恢復)

那麼，初始化的時候，是按照上述的說明進行選舉的，但是當zookeeper運行了一段時間之後，有機器down掉，重新選舉時，選舉過程就相對複雜了。需要加入數據id、leader id和邏輯時鐘。數據id：數據新的id就大，數據每次更新都會更新id。Leader id：就是我們配置的myid中的值，每個機器一個。邏輯時鐘：這個值從0開始遞增,每次選舉對應一個值,也就是說: 如果在同一次選舉中,那麼這個值應該是一致的 ; 邏輯時鐘值越大,說明這一次選舉leader的進程更新.選舉的標準就變成：1、邏輯時鐘小的選舉結果被忽略，重新投票2、統一邏輯時鐘后，數據id大的勝出3、數據id相同的情況下，leader id大的勝出根據這個規則選出leader。

Advertisements

你可能會喜歡

Advertisements