概述
1.zookeeper实现选主从的原理
2.zookeeper实现选主从代码
选主从的原理
在分布式场景中经常会用到zookeeper，常用的有利用zookeeper来选举主从，管理节点状态，或者使用zookeeper来实现分布式锁；具体原理是什么呢？
这里只将实现方式的一种，根据编号大小来实现：（其他方式有通过创建节点实现的，等等）
所有的节点向zk的某个路径下注册，创建临时节点（临时节点，zookeeper会主动监控，一旦连接失效，zk会删除该临时节点），每个注册者创建时会有一个编号，每次选举编号最小的为主节点，其他节点就为从节点，从节点会监控主节点是否失效（怎么监控？ zk有事件，监听事件的状态变化，然后重新选举），为避免“惊群”现象，每个节点只监控比它小的一个临近节点。
代码实现
选主从代码：
import org.apache.zookeeper.*; import org.apache.zookeeper.data.Stat; import java.io.IOException; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; /** * Created by on 17/11/23. */ public class ChooseMaster implements Watcher { private ZooKeeper zk=null; private String selfPath=null; private String waitPath=null; private static final String ZK_ROOT_PATH="/zkmaster"; //选主从的跟路径 private static final String ZK_SUB_PATH=ZK_ROOT_PATH+"/register"; private CountDownLatch successCountDownLatch=new CountDownLatch(1); private CountDownLatch threadCompleteLatch=null; public ChooseMaster(CountDownLatch countDownLatch){ this.threadCompleteLatch=countDownLatch; } @Override public void process(WatchedEvent watchedEvent) { //监听事件 Event.KeeperState keeperState=watchedEvent.getState(); Event.EventType eventType=watchedEvent.getType(); if(Event.KeeperState.SyncConnected==keeperState){ //建立连接 if(Event.EventType.None==eventType){ System.out.println(Thread.currentThread().getName()+" connected to server"); successCountDownLatch.countDown(); }else if(Event.EventType.NodeDeleted==eventType && watchedEvent.getPath().equals(waitPath)){ //监测到节点删除，且为当前线程的等待节点 System.out.println(Thread.currentThread().getName() + " some node was deleted,I'll check if I am the minimum node"); try{ if(checkMinPath()){ //判断自己是不是最小的编号 processMasterEvent(); //处理主节点做的事情 } }catch (Exception e){ e.printStackTrace(); } } }else if(Event.KeeperState.Disconnected==keeperState){ //连接断开 System.out.println(Thread.currentThread().getName()+ " release connection"); }else if(Event.KeeperState.Expired==keeperState){ //超时 System.out.println(Thread.currentThread().getName()+ " connection expire"); } } public void chooseMaster() throws Exception { selfPath=zk.create(ZK_SUB_PATH,null,ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); //创建临时节点 System.out.println(Thread.currentThread().getName()+ "create path "+selfPath); if(checkMinPath()){ //判断是否为主节点 processMasterEvent(); } } public boolean createPersistPath(String path,String data,boolean needWatch) throws KeeperException, InterruptedException { if(zk.exists(path,needWatch)==null){ zk.create(path,data.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); System.out.println(Thread.currentThread().getName()+" create persist path "+path); } return true; } public void createConnection(String connection,int timeout) throws IOException, InterruptedException { zk=new ZooKeeper(connection,timeout,this); successCountDownLatch.await(); } private void processMasterEvent() throws KeeperException, InterruptedException { if(zk.exists(selfPath,false)==null){ System.out.println(Thread.currentThread().getName()+ " selfnode is not exist "+ selfPath); return; } System.out.println(Thread.currentThread().getName()+ " I'm the master,now do work"); Thread.sleep(2000); System.out.println(Thread.currentThread().getName()+" Finish do work,leave master"); //zk.delete(selfPath,-1); releaseConnection(); threadCompleteLatch.countDown(); } private void releaseConnection() { if(zk!=null){ try { zk.close(); } catch (InterruptedException e) { e.printStackTrace(); } } } private boolean checkMinPath() throws Exception {
//获取根节点下的所有子节点，进行排序，取当前路径的index，如果排在第一个，则为主，否则检测前一个节点是否存在，不存在则重新选举最小的节点 List<String> subNodes=zk.getChildren(ZK_ROOT_PATH,false); System.out.println(subNodes.toString()); Collections.sort(subNodes); System.out.println(Thread.currentThread().getName()+" tmp node index is "+selfPath.substring(ZK_ROOT_PATH.length()+1)); int index=subNodes.indexOf(selfPath.substring(ZK_ROOT_PATH.length()+1)); switch (index){ case -1: System.out.println(Thread.currentThread().getName()+" create node is not exist"); return false; case 0: System.out.println(Thread.currentThread().getName()+" I'm the master"); return true; default: waitPath=ZK_ROOT_PATH+"/"+subNodes.get(index-1); System.out.println(Thread.currentThread().getName()+" the node before me is "+waitPath); try{ zk.getData(waitPath,true,new Stat()); return false; }catch (Exception e){ if(zk.exists(waitPath,false)==null){ System.out.println(Thread.currentThread().getName()+" the node before me is not exist,now is me"); return checkMinPath(); }else{ throw e; } } } } }
测试代码：
import org.apache.zookeeper.KeeperException; import java.io.IOException; import java.util.concurrent.*; /** * Created by on 17/11/23. */ public class MasterChoiceTest { private final static String ZK_CONNECT_STRING="127.0.0.1:2181"; private final static String ZK_ROOT_PATH="/zkmaster"; private final static int SESSION_TIMEOUT=10000; private static final int THREAD_NUM=5; private static int threadNo=0; private static ExecutorService executorService=null; private static CountDownLatch threadCompleteLatch=new CountDownLatch(THREAD_NUM); public static void main(String[] args){ executorService= Executors.newFixedThreadPool(THREAD_NUM, new ThreadFactory() { @Override public Thread newThread(Runnable r) { String name=String.format("The %s thread",++threadNo); Thread ret=new Thread(Thread.currentThread().getThreadGroup(),r,name,0); ret.setDaemon(false); return ret; } }); if(executorService!=null){ startProcess(); } } private static void startProcess() { Runnable masterChoiceTest=new Runnable() { @Override public void run() { String threadName=Thread.currentThread().getName(); ChooseMaster chooseMaster=new ChooseMaster(threadCompleteLatch); try { chooseMaster.createConnection(ZK_CONNECT_STRING,SESSION_TIMEOUT); System.out.println(Thread.currentThread().getName()+" connected to server"); synchronized (MasterChoiceTest.class){ chooseMaster.createPersistPath(ZK_ROOT_PATH,"thread "+threadName,true); } chooseMaster.chooseMaster(); } catch (IOException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } catch (Exception e) { e.printStackTrace(); } } }; for(int i=0;i<THREAD_NUM;i++){ executorService.execute(masterChoiceTest); } executorService.shutdown(); try { threadCompleteLatch.await(); System.out.println("All thread finished"); } catch (InterruptedException e) { e.printStackTrace(); } } }
输出结果：
The 1 thread-EventThread connected to server
The 5 thread-EventThread connected to server
The 3 thread-EventThread connected to server
The 2 thread-EventThread connected to server
The 5 thread connected to server
The 3 thread connected to server
The 2 thread connected to server
The 1 thread connected to server
The 4 thread-EventThread connected to server
The 4 thread connected to server
The 5 thread create persist path /zkmaster
The 5 threadcreate path /zkmaster/register0000000000
[register0000000000]
The 5 thread tmp node index is register0000000000
The 5 thread I'm the master
The 1 threadcreate path /zkmaster/register0000000001
The 2 threadcreate path /zkmaster/register0000000002
The 5 thread I'm the master,now do work
The 3 threadcreate path /zkmaster/register0000000003
[register0000000000, register0000000002, register0000000001]
The 1 thread tmp node index is register0000000001
The 1 thread the node before me is /zkmaster/register0000000000
[register0000000000, register0000000002, register0000000001, register0000000003]
[register0000000000, register0000000002, register0000000001, register0000000003]
The 3 thread tmp node index is register0000000003
The 2 thread tmp node index is register0000000002
The 3 thread the node before me is /zkmaster/register0000000002
The 2 thread the node before me is /zkmaster/register0000000001
The 4 threadcreate path /zkmaster/register0000000004
[register0000000000, register0000000002, register0000000001, register0000000004, register0000000003]
The 4 thread tmp node index is register0000000004
The 4 thread the node before me is /zkmaster/register0000000003
The 5 thread Finish do work,leave master
The 1 thread-EventThread some node was deleted,I'll check if I am the minimum node
[register0000000002, register0000000001, register0000000004, register0000000003]
The 1 thread-EventThread tmp node index is register0000000001
The 1 thread-EventThread I'm the master
The 1 thread-EventThread I'm the master,now do work
The 1 thread-EventThread Finish do work,leave master
The 2 thread-EventThread some node was deleted,I'll check if I am the minimum node
[register0000000002, register0000000004, register0000000003]
The 2 thread-EventThread tmp node index is register0000000002
The 2 thread-EventThread I'm the master
The 2 thread-EventThread I'm the master,now do work
The 2 thread-EventThread Finish do work,leave master
The 3 thread-EventThread some node was deleted,I'll check if I am the minimum node
[register0000000004, register0000000003]
The 3 thread-EventThread tmp node index is register0000000003
The 3 thread-EventThread I'm the master
The 3 thread-EventThread I'm the master,now do work
The 3 thread-EventThread Finish do work,leave master
The 4 thread-EventThread some node was deleted,I'll check if I am the minimum node
[register0000000004]
The 4 thread-EventThread tmp node index is register0000000004
The 4 thread-EventThread I'm the master
The 4 thread-EventThread I'm the master,now do work
The 4 thread-EventThread Finish do work,leave master
All thread finished
分布式锁的原理也是一样，每次编号最小的获取锁。