zookeeper学习笔记二

这几天尝试进行leader/follower的测试,当然目前还没去看源码实现.但觉得还是相当的.

以下是对网上一段代码的修改,因为原来的测不出来什么东西.

note:

运行时只有两个节点时(leader&follower),当leader down后,the other one 一般很少自动切换到ld;但三个以上是没有问题的,何解?不知道是不是所谓的'ensemble'模式下'大部分机器正常动作才提供supply services这个原因呢...

 

 
/**
 * LeaderElection
 * 通用实现方案:(MY)
 * 假如这个最小编号的 Server 死去，**由于是 EPHEMERAL 节点，死去的 Server 对应的节点也被删除**，
 * 所以当前的节点列表中又出现一个最小编号的节点，我们就选择这个节点为当前 Master。
 * 这样就实现了动态选择 Master，避免了传统意义上单 Master 容易出现单点故障的问题。(可以参考本项目Locks.java)
 * NOTE:这里的实现是简化的
 * 
 * 本测试需要先单独启动TestMainServer(也可以打开TestMainServer.start()再启clients RAW),
 *  然后再启动其它若干个,
 *  再手工删除master node:/GroupMembers/leader,
 * 观察其它节目是否可以顺利过渡到新的leader-followers状态.

 */
public class LeaderElection extends TestMainClient {
    public static final Logger logger = Logger.getLogger(LeaderElection.class);
 
    public LeaderElection(String connectString, String root) {
        super(connectString);
//        try {
//            connectedSignal.await();    //MY test result:no effect
//        } catch (InterruptedException e1) {
//            e1.printStackTrace();
//        }     
         
        this.root = root;
        if (zk != null) {
            try {
                Stat s = zk.exists(root, false);
                if (s == null) {
                    zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                }
            } catch (KeeperException e) {
                logger.error(e);
            } catch (InterruptedException e) {
                logger.error(e);
            }
        }
    }
 
    void findLeader() throws InterruptedException, UnknownHostException, KeeperException {
        byte[] leader = null;
        try {
            leader = zk.getData(root + "/leader", true, null);
        } catch (KeeperException e) {
            if (e instanceof KeeperException.NoNodeException) {    //this exception is acceptable
                logger.error(e);
                e.printStackTrace();
            } else {
                throw e;
            }
        }
        //已经存在master,只能作为follower
        if (leader != null) {
            System.out.println("10");
            following();
            //MY
            synchronized(mutex){
                mutex.wait();
            }
            System.out.println("14");
            findLeader();    //继续竟争master
        } else {
            System.out.println("11");
            String newLeader = null;
            byte[] localhost = InetAddress.getLocalHost().getAddress();
            try {
                newLeader = zk.create(root + "/leader", localhost, ZooDefs.Ids.OPEN_ACL_UNSAFE, 
                        CreateMode.EPHEMERAL);    //NOTE: 必须为临时节点,server死亡后可以由其它节点实时选举leader
                zk.exists(root + "/leader", true);    //set a watch for the COMING leader!
                
            } catch (KeeperException e) {
                //并发运行时可能出现此情况,this exception is acceptable
                if (e instanceof KeeperException.NodeExistsException) {
                    logger.error(e);
                    e.printStackTrace();
                } else {
                    throw e;
                }
            }
            if (newLeader != null) {
                leading();
                //other things to process...
                //for example
              //MY 驻留
                System.out.println("1");
                synchronized(mutex){
                    try {
                        System.out.println("2");
                        mutex.wait();
                        System.out.println("3");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                //完成任务后放弃leader权限,但继续竟选leader
                
                //模拟死亡
                System.out.println("4");
                System.exit(1);        //EXIT
                System.out.println("5");
            } else {
                System.out.println("12");
                synchronized(mutex){
                    mutex.wait();
                }
                System.out.println("13");
                //MY
                findLeader();////继续竟争master
            }
        }
    }
 
    @Override
    public void process(WatchedEvent event) {
        System.out.println("9");
        if(event.getState() == KeeperState.SyncConnected){
            //MY
            //connectedSignal.countDown();
        }
         
        if (event.getPath().equals(root + "/leader") ) {
            if(event.getType() == Event.EventType.NodeCreated){
                System.out.println("得到通知-leader NodeCreated");
                //RAW
//                super.process(event);
//                following();
                
            }
            //MY 删除master时通知其它followers竟争选择新master
            if(event.getType() == Event.EventType.NodeDeleted){
                System.out.println("6");
                synchronized(mutex){
                    //System.out.println("7");
                    mutex.notify();    //如果每个应用中只存在一个waiter则可以使用这方法,否则使用notifyAll()
                } 
                System.out.println("8");
                //MY
                //模拟:如果是本server宕机了,删除master node(如果没删除),让其它nodes选举一个继任master
//                try {
//                    byte[] localhost = InetAddress.getLocalHost().getAddress();
//                    byte[] leader = zk.getData(root + "/leader", true, null);
//                    if(new String(localhost).equals(new String(leader))){
//                        zk.delete(root + "/leader", -1);
//                        //退出本应用.所以本应用最好使用TestMainServer先运行提供服务,
//                        //而不是利用一个启动LeaderElection驻留来提供服务
//                    }else{
//                        //do nothing,其它node继续寻求继任master
//                    }
//                } catch (UnknownHostException e) {
//                    e.printStackTrace();
//                } catch (KeeperException e) {
//                    e.printStackTrace();
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
            }
        }
    }
 
    void leading() {
        System.out.println("成为领导者");
    }
 
    void following() {
        System.out.println("成为组成员");
    }
 
    public static void main(String[] args) {
//        TestMainServer.start();        //if enable this clause,the flag EXIT must be commented
        String connectString = "localhost:" + TestMainServer.CLIENT_PORT;
 
        LeaderElection le = new LeaderElection(connectString, "/GroupMembers");
        try {
            le.findLeader();
        } catch (Exception e) {
            logger.error(e);
            e.printStackTrace();
        }
    }
}

 

 

tickTime：这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。
Zookeeper 还支持另外一种伪集群的方式，也就是可以在一台物理机上运行多个 Zookeeper 实例(hbase也

集群模式配置（比单机模式多添加的项)
 initLimit=5
 syncLimit=2
 server.1=192.168.211.1:2888:3888
 server.2=192.168.211.2:2888:3888
* initLimit：这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 10 个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 5*2000=10 秒
* syncLimit：这个配置项标识 Leader 与 Follower 之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是 2*2000=4 秒
* server.A=B：C：D：其中 A 是一个数字，表示这个是第几号服务器；B 是这个服务器的 ip 地址；C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；D 表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。如果是伪集群的配置方式，由于 B 都是一样，所以不同的 Zookeeper 实例通信端口号不能一样，所以要给它们分配不同的端口号。

利用zkCli.sh -server <host>:<port>可以连接指定的zk Servers;
不同的host或port建立的nodes,是存放不同的.

zk.setData()方法后，watch恢复false默认
是否产生event只决定于最后一次的设置true/false