文档章节

Zookeeper【V3.4.6】-Programing Guide 2:barries

止静
 止静
发布于 2014/11/04 16:18
字数 1082
阅读 63
收藏 1


1: Barries:  栅栏,见面知意。

2:Queue:Queue也就是我们所说的队列



1:Barries:

            1.1: 是指所有的现场都达到 barrier后才能进行后续的计算

            1.2:所有的线程都完成自己的计算以后才能离开barrier            

        

进入栅栏:
        1,新建一个根节点 "/root"
        2, 想进入barrier的线程在 “/root”下建立一个字节点"/root/c-i"
        3,循环监听"/root"孩子节点数的变化,每当其达到Size的时候就说明有Size个线程都已经达到了Barrier的
要求。

  2:Queue:就是指一个生产者或消费者的模型 

离开Barrier
        1:  想离开Barrier的现场删除掉在"/root" 下建立的子节点
        2:  循环监听"/root" 孩子节点数目的变化,当Size减少到0的时候它就可以离开了。

 3 :Queue 队列的实现 

    

1 : 建立一个根节点"/root"
2  : 生产线程在"/root" 下建立一个SEQUENTAIL的节点
3 : 消费线程检查"/root" 如果没有就循环的监听"/root" 节点的变化,直到它有自己的子节点,删除序号
最小子字节点。

package sync;
 
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Random;
 
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
 
public class SyncPrimitive implements Watcher {
    static ZooKeeper zk = null;
    static Integer mutex;
    String root;
 
    //同步原语
    SyncPrimitive(String address) {
        if (zk == null) {
            try {
                System.out.println("Starting ZK:");
                //建立Zookeeper连接,并且指定watcher
                zk = new ZooKeeper(address, 3000, this);
                //初始化锁对象
                mutex = new Integer(-1);
                System.out.println("Finished starting ZK:" + zk);
            } catch (IOException e) {
                System.out.println(e.toString());
                zk = null;
            }
        }
    }
 
    @Override
    synchronized public void process(WatchedEvent event) {
        synchronized (mutex) {
            //有事件发生时,调用notify,使其他wait()点得以继续
            mutex.notify();
        }
 
    }
 
    static public class Barrier extends SyncPrimitive {
        int size;
        String name;
 
        Barrier(String address, String root, int size) {
            super(address);
            this.root = root;
            this.size = size;
            if (zk != null) {
                try {
                    //一个barrier建立一个根目录
                    Stat s = zk.exists(root, false);        //不注册watcher
                    if (s == null) {
                        zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
                                CreateMode.PERSISTENT);
                    }
                } catch (KeeperException e) {
                    System.out
                            .println("keeper exception when instantiating queue:"
                                    + e.toString());
                } catch (InterruptedException e) {
                    System.out.println("Interrupted exception.");
                }
            }
            try {
                //获取自己的主机名
                name = new String(InetAddress.getLocalHost()
                        .getCanonicalHostName().toString());
            } catch (UnknownHostException e) {
                System.out.println(e.toString());
            }
        }
 
        boolean enter() throws KeeperException, InterruptedException {
            //在根目录下创建一个子节点.create和delete都会触发children wathes,这样getChildren就会收到通知,process()就会被调用
            zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);
            //一直等,直到根目录下的子节点数目达到size时,函数退出
            while (true) {
                synchronized (mutex) {
                    List<String> list = zk.getChildren(root, true);
                    if (list.size() < size) {
                        mutex.wait();       //释放mutex上的锁
                    } else {
                        return true;
                    }
                }
            }
        }
 
        boolean leave() throws KeeperException, InterruptedException {
            //删除自己创建的节点
            zk.delete(root + "/" + name, 0);
            //一直等,直到根目录下有子节点时,函数退出
            while (true) {
                synchronized (mutex) {
                    List<String> list = zk.getChildren(root, true);
                    if (list.size() > 0) {
                        mutex.wait();
                    } else {
                        return true;
                    }
                }
            }
        }
 
    }
 
    static public class Queue extends SyncPrimitive {
        Queue(String address, String name) {
            super(address);
            this.root = name;
            if (zk != null) {
                try {
                    //一个queue建立一个根目录
                    Stat s = zk.exists(root, false);
                    if (s == null) {
                        zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
                                CreateMode.PERSISTENT);
                    }
                } catch (KeeperException e) {
                    System.out
                            .println("keeper exception when instantiating queue:"
                                    + e.toString());
                } catch (InterruptedException e) {
                    System.out.println("Interrupted exception.");
                }
            }
        }
 
        //参数i是要创建节点的data
        boolean produce(int i) throws KeeperException, InterruptedException {
            ByteBuffer b = ByteBuffer.allocate(4);
            byte[] value;
            b.putInt(i);
            value = b.array();
 
            //根目录下创建一个子节点,因为是SEQUENTIAL的,所以先创建的节点具有较小的序号
            zk.create(root + "/element", value, Ids.OPEN_ACL_UNSAFE,
                    CreateMode.PERSISTENT_SEQUENTIAL);
            return true;
        }
 
        int consume() throws KeeperException, InterruptedException {
            int retvalue = -1;
            Stat stat = null;
            while (true) {
                synchronized (mutex) {
                    List<String> list = zk.getChildren(root, true);       //并不能保证list[0]就是序号最小的
                    //如果根目录下没有子节点就一直等
                    if (list.size() == 0) {
                        System.out.println("Going to wait");
                        mutex.wait();
                    }
                    //找到序号最小的节点将其删除
                    else {
                        Integer min = new Integer(list.get(0).substring(7));
                        for (String s : list) {
                            Integer tmp = new Integer(s.substring(7));
                            if (tmp < min)
                                min = tmp;
                        }
                        System.out.println("Temporary value:" + root
                                + "/element" + min);
                        byte[] b = zk.getData(root + "/element" + min, false,
                                stat);
                        zk.delete(root + "/element" + min, 0);
                        ByteBuffer buffer = ByteBuffer.wrap(b);
                        retvalue = buffer.getInt();
                        return retvalue;
                    }
                }
            }
        }
    }
 
    public static void main(String[] args) {
        if (args[0].equals("qTest"))
            queueTest(args);
        else
            barrierTest(args);
    }
 
    private static void barrierTest(String[] args) {
        Barrier b = new Barrier(args[1], "/b1", new Integer(args[2]));
        try {
            boolean flag = b.enter();
            System.out.println("Enter barrier:" + args[2]);
            if (!flag)
                System.out.println("Error when entering the barrier");
        } catch (KeeperException e) {
 
        } catch (InterruptedException e) {
        }
 
        Random rand = new Random();
        int r = rand.nextInt(100);
        for (int i = 0; i < r; i++) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
 
            }
        }
        try {
            b.leave();
        } catch (KeeperException e) {
 
        } catch (InterruptedException e) {
        }
        System.out.println("Left barrier");
    }
 
    private static void queueTest(String[] args) {
        Queue q = new Queue(args[1], "/app1");
        System.out.println("Input:" + args[1]);
        int i;
        Integer max = new Integer(args[2]);
 
        if (args[3].equals("p")) {
            System.out.println("Producer");
            for (i = 0; i < max; i++)
                try {
                    q.produce(10 + 1);
                } catch (KeeperException e) {
 
                } catch (InterruptedException e) {
                }
        } else {
            System.out.println("Consumer");
            for (i = 0; i < max; i++)
                try {
                    int r = q.consume();
                    System.out.println("Item:" + r);
                } catch (KeeperException e) {
                    i--;
                } catch (InterruptedException e) {
                }
        }
    }
 
}


© 著作权归作者所有

共有 人打赏支持
止静
粉丝 121
博文 134
码字总数 125762
作品 0
东城
技术主管
私信 提问
Zookeeper CancelledKeyException

随着生产集群里应用的增多,zookeeper的压力越来越大,resourcemanager出现了异常挂起。 延迟问题 首先是“fsync-ing the write ahead log in SyncThread:3 took 1606ms which will adversel...

Yulong_
2017/08/10
0
1
Zookeeper 3.4.x安装和配置--Linux篇

阅读目录: 1. 关闭防火墙和Selinux 2. 安装所需环境JDK 3. 下载Zookeeper 3.4.x版本 4. 配置并启动Zookeeper 5. 验证并配置自启动 6. 说明 1. 关闭防火墙和Selinux Linux的防火墙是咱们新手...

loubobooo
2018/02/28
0
0
Kafka与.net core(一)安装

1.安装JDK 目前官网不能直接下载,在网上找到1.8.0版本安装包下载到本地。 1.1.下载jdk并解压 [root@iz2zei2y693gtrgwlibzlwz java]# lsjdk1.8.0_191 jdk-8u191-linux-x64.tar.gz 1.2.配置j...

老六代码
01/10
0
0
zookeeper 和 kafka 的安装使用

前提都安装JDK 6版本以上 java -version java version "1.7.0_79" 测试环境集群结构(三台集群的hosts一定要互相解析 不然会很坑) server1: 192.168.100.10 server1: 192.168.100.13 server1......

cuizhiliang
2018/06/26
0
0
2018-07-11期 Zookeeper集群安装部署【本人亲自反复验证通过分享】

简介: ZooKeeper是一个分布式的,开放源码的分布式应用程序协调服务,是Google的Chubby一个开源的实现,是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件,提供的功能...

JackmaSong
2018/07/11
0
0

没有更多内容

加载失败,请刷新页面

加载更多

安卓constraintLayout中app:srcCompat设置的图片显示不出来

使用 app:srcCompat 的时候 引入的图片显示不出来的解决方案 首先查看的你的Activity 继承的是那个Activity 如果是继承AppcompatActivity 使用 ImageView的 app:srcCompat 是没有问题的 如果...

雨焰
40分钟前
1
0
MySQL mysqldump数据导出详解

MySQL mysqldump数据导出详解 2016-04-07 11:14 by pursuer.chen, 114348 阅读, 0 评论, 收藏, 编辑 介绍 在日常维护工作当中经常会需要对数据进行导出操作,而mysqldump是导出数据过程中使用...

linjin200
40分钟前
0
0
Lucene 检索时的步骤

用户输入查询语句:lucene AND learned NOT hadoop 搜索主要分为以下几步 第一步、对查询语句进行词法分析,语法分析,及语言处理 词法分析主要用来识别单词和关键字 语法分析主要是根据查询...

kdy1994
41分钟前
2
0
Gradle Implementation vs API configuration

注: 可以把 Implementation理解为java类的private, 使用Implementation则外部依赖本库的代码引用不到本库Implementation的代码, 也就是降低依赖树的层级, 这样就可以避免外部代码编译时候, 导...

SuShine
45分钟前
2
0
pdf安卓手机适配以及ie的兼容 以及隐藏掉下载打印按钮的方式

这是我目前遇到pdf方面最多的问题。 在百度上搜了很多答案,结果都是建议使用插件。而目前百度能搜的插件基本上都是有缺陷的。自己测试了很多插件。最后使用了pdf.js这个插件。这个插件目前完...

流年那么伤
51分钟前
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部