zookeeper怎么分发任务的
发布网友
发布时间：2022-04-24 23:28
共1个回答
热心网友
时间：2023-10-14 12:11
zookeeper要结合分布式内存存储工具（像memcached、redis、MQ工具等），能发挥很强大的分布式计算功能。有些很强大的工具都用到zookeeper了，像hadoop、storm、阿里bbo等等。
这里基zookeeper的选举，没有那么复杂，抛砖引玉了
2014－04－27：经过一个项目的熏陶，对zookeeper有了一定的理解，Paxos是zookeeper的核心算法，我们是在zookeeper基础上做开发，算法理解就行了，但别给算法给绑死了，没有哪个程序要求一定要实现什么算法，必须知道zookeeper的作用是协调分布式应用。
/**
*
* @author lyq
*
*/
public class ZkAppCenter implements Watcher {
//private static ZkAppInfo info = null;
public static final String appGroup = "/APP";
private String hosts = BaseCode.getValue("ZK_HOST");
private Integer SESSION_TIMEOUT = Integer.parseInt(BaseCode.getValue("ZK_SESSION_TIMEOUT"));
String taskPath = appGroup + "_leader";
public CountDownLatch lockLatch = null;
private ZooKeeper zk = null;
protected static final Logger log = Logger.getLogger(ZkAppCenter.class);
private String appName = null;
//public static ZkAppInfo getInstance(){
//if(info == null){
//info = new ZkAppInfo();
//}
//return info;
//}
private ZooKeeper getZK() throws Exception{
if(zk==null){
zk = new ZooKeeper(hosts, SESSION_TIMEOUT,this);
}
return zk;
}
/**
*
* @return
* @throws Exception
*/
public String registerApp() throws Exception{
if(getZK().exists(appGroup, false)==null){
getZK().create(appGroup,appGroup.getBytes(),Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
ZkLock lock = new ZkLock(appGroup+"lock",this.getZK()) {//创建应用构建必须是有序的，锁在我上篇文字
@Override
public Object doAction() {
try {
String addr = InetAddress.getLocalHost().getHostAddress();
appName = zk.create(appGroup+"/"+addr+"-",new byte[0] , Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
if(zk.exists(taskPath, false)==null){
zk.create(taskPath, "leader".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);//创建任务发布路径
}
zk.getData(taskPath,true, null);
//listenChain();
//listenLeader();//监听leader节点
return null;
} catch (Exception e) {
throw new RuntimeException(e);
}//每个应用注册进来必须监听leader创建的leader路径
}
};
lockLatch = lock.latch;
lock.execute();
return appName;
}
//发布应用任务时，只需要修改taskPath，因为所有应用都有watch taskPath，所以所有应用都执行了该任务；（任务集合可以放在集群内存队列里面，让每个应用去拿）
public void doTask(String taskName) throws Exception{
ZkTask taskBean = ZkTaskFactory.getInstance().getTask(taskName);
Object obj = taskBean.execute();
if(obj!=null){
this.getZK().setData(taskPath, taskName.getBytes(), -1);
}else{
this.getZK().setData(taskPath,taskName+":end",-1);//结束命令
}
}
public Integer getAppsSize() throws Exception{
List<String> list = this.getZK().getChildren(appGroup, false);
Integer appsSize = list==null?0:list.size();
if(appsSize==0){
log.error("集群失败，没有'"+appGroup+"'该应用群组里。");
}
return appsSize;
}
public static void main(String[] args) throws Exception {
try{
final CountDownLatch latchtask = new CountDownLatch(10);
for(int i = 0;i<10;i++){
final int ia = i;
new Thread(){
public void run(){
ZkAppCenter reg = new ZkAppCenter();
try {
reg.registerApp();
latchtask.await();
Thread.sleep(4000);
if(ia==9){
List<ZkTaskBean> list = new ArrayList();
for(int i=0;i<8000;i++){
list.add(new DoloadTaskBean());
}
new ZkPlan(BaseCode.RD_DOLOAD_TASKS,list,reg.getAppsSize(), 1000);
reg.doTask("TestTask");
}
} catch (Exception e) {
e.printStackTrace();
}
}
}.start();
latchtask.countDown();
}
Thread.sleep(Long.MAX_VALUE);
}finally{
}
}
//2014-04-27 lyq 这里有个问题，如果process执行太久，会导致其他任务发布不了，估计是加锁了，所以process方法中执行任务时，最好开启一个子线程：
new Thread(){
public void run() {
try {
doTask(task);
} catch (Exception e) {
e.printStackTrace();
}
}
}.start();
这样各种类型任务互不干扰，项目实战经验啊。其实想过为每个任务建zookeeper连接，但是zookeeper默认就10个连接，建太多连接还要抽象出代码去同意管理连接，因为连接是长连接，如果不抽象这块连接的程序，那么当程序挂了要费很大的劲去为每个连接建立容灾的程序；还是单个连接的路线吧。还有一些容灾的方法，跟本博客redis做法是一样的：如果失去连接，让所有任务进入等待，关闭zookeeper，重新在zookeeper注册应用。再让zookeeper任务继续执行，这里在getZk()这个方法放入监听锁~~
多线程和分布式的应用，可以充分体验锁的魅力。
zookeeper真心不错～～～
@Override
public void process(WatchedEvent event) {
try {
String eventPath = event.getPath();
if(eventPath!=null&&eventPath.contains("lock")){
this.lockLatch.countDown();//锁
}else if (event.getType() == EventType.None && event.getState() == KeeperState.SyncConnected) {
String addr = InetAddress.getLocalHost().getHostAddress();
log.info(addr + " register......");
}else if(event.getType() == EventType.NodeDeleted){
log.info(eventPath+"节点删除..........");
if(eventPath.equals(this.taskPath)){
this.registerApp();//监听到父节点删除，那么重新触发注册，防止领导节点挂掉、
}
}else{
String task = new String(this.getZK().getData(eventPath, true, null));//每次监听到任务，继续监听
doTask(task);
//}
this.log.info("this.appName:"+this.appName);
//
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
