zk master-slaver机制

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

1、基本概念

>>zookeeper handler （zk句柄）有点类似文件句柄，打开一个文件就保持了一个文件句柄！同样的道理：建立一个到zk server的session就会有一个zk handler（可以用任何编程语言）。

>> move session：client 可以 connect到任意而且只能是一个server，但是如果超时（好久没有从server上hearded了）这个session可以迁移到其他server上，叫做 moving session

>> 一个client可以有多个session

2、master-worker 例子 zkcli 版本

2.1、Master:

watch for new workers and tasks,and assign tasks to workers;//其实这里做了3件事，而且是有顺序的：先warch for availabe worker 然后 watch for tasks 然后 assign

>>一个系统只有一个process可以成为master，一个进程为了成为master 必须创建一个 ephemeral的znode call /master

 create -e /master "host:port" //-e表示znode模式为 ephemeral  这里的数据 host:port 目的是让别人可以在zk之外访问他 ephemeral表示如果session失效 znode删除

 ls /                                          //list the root of zk tree

 get /master                            //get metadata

//watch 伴随着znode，如果znode发生变化 就会有事件
//stat 作用两个1 get the attributes of a znode；2 才存在的znode上设置watch
4 stat /master true                       // 参数 为 紧跟在path后的true表示sets the watch

然后master需要创建一些parent 的 znode，然后master需要watch哪个worker是可用的？那些task要分配？然后做分配

//是persistent znode不包含数据，作用就是告诉哪个worker是可用的？何时有task可以分配？然后分配给workers

create /worker ""

create /tasks ""

create /assigns ""

ls /

//Master 需要watch for changes in the childer of /workers and /tasks，参数true表示为znode创建了watch

ls /worker true

ls /tasks true

2.2、worker:

将自己注册到系统，对master表明自己可用，这样话 master就可以watch for new tasks

//worker第一步就是通知master 它存在了可以执行taskle

//a、在/workers 创建一个ephemeral znode

create -e /worker/worker1 "host:port"

//因为master在关注 /workers所以一旦在worker1这个znode创建了，master就会观察到了 通知

WATCHER::

WatchedEvent state:SyncConneted type:NodeChildreChanged paht:/workders

//b、然后创建一个parent znode /assign/worker1 to receive assignments,and watch it for new tasks by ls true

create -e /assign/worker1

ls /assign/worker1 true

到此 worker准备接受assign了

//c、当得到master 通知 a new task has been assigned,worker 就会check new task and 确保 task 被assigned给了自己

ld /assign/worker1

//d、如果执行完了以后  就会在znode /tasks创建一个status znode

create /tasks/task-00000000/status "done"

2.3、client：

create new tasks 然后等待系统反应

这里简化下任务就是client要求执行cmd命令

1、首先 to add a task

zkcli>create -s /tasks/task-  "cmd"   //mode是sequential，用户就会等待task被执行了

Created /tasks/task-0000000

2、执行这个task的worker执行完毕这个task后会 create a status znode for the task注意创建目录是在 /tasks/task-00000000,所以如过client看到这个status znode创建就知道task执行完了，所以他一直watch for the creation of the status znode.....

ls /tasks/task-0000000 true

3、master看到一个new task就会 get availabe workers ,然后assign

ls /tasks

ls /workers

create /assign/worker1/task-000000 ""

3、zk API

zk handler represents a session wieth zk,如果zk handler is colosed，zk client library will tell zk servers to kill the session. client 利用zk handler 连接zk server 。。。。

》》创建一个zk handler

Zookeeper(String connectString,int sesssionTimeout,Watcher watcher);

注意Watcher是一个interface，你需要implements然后实例化，这个warcher关注 session的event(比如建立或者关闭)或者 changes of zk data

实现一个Watcher

import org.apache.zookeeper.ZooKeeper;

import org.apache.zookeeper.Watcher;

public class Master implements Watcher{

//这个serverId是创建znode的进程产生的一个随机数

private statci String serverId=Integer.toHexString(random.nextInt());

private isLeader=false;//当前进程是否为master进程？？？？

 ZooKeeper zk;

String hostPort;

void startZK(){

//注意：must consturt the zk object after Master的构造函数执行完，

 zk=new ZooKeeper(hostPort,15000,this);

}

//覆盖的方法

public void process(WatchEvent e){

System.out.println(e);

}

/**

public byte[] getData(String path,bool watch,Stat stat);

//注意参数watch：if set true,将会得到events on the Warcher object we set when we created the zk handler.

//这里设置为false表示只关注 当前的数据是啥？？

//stat参数是当前方法可以对znode填充的元数据

//返回值 如果成功返回 the data of znode

//如下函数判断是否已经存在master了？？？

**/

//判断当前zk集群知否存在了master？？？？

public boolean checkMaster(){

 try{

Stat stat=new Stat();

byte data[]=zk.getData("/master",false,stat);

isLeader=new String(data).equals(serverId);

return true;

}catch(NoNodeException e){

return false;

}

}

//创建znode znode的初始化数据必须和创建这个znode的进程有关系（这里是每个进程产生一个random 的serverid），然后还要注意权限

public void runForMaster(){

 try{

//创建znode

zk.create("/master",serverId.getBytes(),OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);

isLeader=true;

}catch(NodeExistsException e){

isLeader=false;

}catch(ConnectionLossException e){

// when a client becomes disconnected from a zk server,可能的原因是 网络error或者 zk server的错误

//这个时候client 不知道当前请求 zk server是否做了处理？？

}

}

//

void stopZK() threos Exception{

zk.close();

}

public static void main(String args[]){

Master m=new Master(args[0]);

m.startZK();

//注意：一旦连接上zk，会有一个后台线程维护zk session，

Thread.sleep(60000);

m.stopZK();

}

}