分布式协调服务zookeeper总结

存储架构 2017-12-18

Zookeeper是一个分布式协调服务,换言之,就是为用户的分布式应用程序提供协调服务

  • zookeeper是为别的分布式程序服务的
  • Zookeeper本身就是一个分布式程序(只要有半数以上节点存活,zk就能正常服务)
  • Zookeeper所提供的服务涵盖:主从协调、服务器节点动态上下线、统一配置管理、分布式共享锁、统一名称服务
  • 虽然说可以提供各种服务,但是zookeeper在底层其实只提供了两个功能(管理数据和监听数据): 管理(存储,读取)用户程序提交的数据;
    并为用户程序提供数据节点监听服务;

1.2 Zookeeper集群的角色: Leader 和 follower

Zookeeper在配置文件中并没有指定master和slave,启动之后通过内部的选举机制选举出leader和follower,而且只有一个leader,其他则为follower。zookeeper集群中只要有半数以上节点存活,集群就能提供服务。 2.zookeeper集群机制 半数机制:集群中半数以上机器存活,集群可用。 zookeeper适合装在奇数台机器上!!!

2.zookeeper安装与配置

2.1zookeeper安装

  • 安装到3台虚拟机上(需要提前安装好JDK) 将zookeeper压缩包上传至/apps/package目录并解压
tar -zxvf zookeeper-3.4.5.tar.gz
  • 重命名
mv zookeeper-3.4.5 zookeeper(重命名文件夹zookeeper-3.4.5为zookeeper)
  • 修改环境变量
vi /etc/profile
添加内容:
export ZOOKEEPER_HOME=/apps/package/zookeeper
export PATH=$PATH:$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin
  • 重新编译文件: source /etc/profile 三台机器都需要修改

2.2 修改配置文件

  • 先复制一份 cd zookeeper/conf cp zoo_sample.cfg zoo.cfg
  • 编辑 vi zoo.cfg
添加内容:
dataDir=/apps/package/zookeeper/data
dataLogDir=/apps/package/zookeeper/log
server.1=mini1:2888:3888
server.2=mini2:2888:3888
server.3=mini3:2888:3888
  • 创建文件夹:
cd /apps/package/zookeeper
mkdir -m 755 data
mkdir -m 755 log
  • 在data文件夹下新建myid文件,myid的文件内容为:
cd data
vi myid
添加内容:
1

mini2和mini3服务器的请修改成2,3,将来会按这个myid选中出leader和follow。

  • 将集群复制到其他机器上
scp -r /apps/package/zookeeper root@mini2:/apps/package/
scp -r /apps/package/zookeeper root@mini3:/apps/package/

如果在mini1中ping不通mini2和mini3,需要在hosts文件中配置mini2和mini3的ip地址

  • 修改其他机器的配置文件 到mini2上:修改myid为:2 到mini3上:修改myid为:3 而且/etc/profile的路径也不要忘了修改
  • 启动(每台机器)
zkServer.sh start
zkServer.sh start-foreground(可以看到启动日志)
  • 查看集群状态
jps(查看进程)
zkServer.sh status(查看集群状态,主从信息)

如果报端口占用,参考下面链接解决:http://blog.csdn.net/u014686180/article/details/51767863

3.zookeeper数据结构和常用操作

3.1zookeeper特性

  • Zookeeper:一个leader,多个follower组成的集群
  • 全局数据一致:每个server保存一份相同的数据副本,client无论连接到哪个server,数据都是一致的
  • 分布式读写,更新请求转发,由leader实施
  • 更新请求顺序进行,来自同一个client的更新请求按其发送顺序依次执行
  • 数据更新原子性,一次数据更新要么成功,要么失败
  • 实时性,在一定时间范围内,client能读到最新数据

3.2zookeeper数据结构

  • 层次化的目录结构;
  • 每个节点在zookeeper中叫做znode,并且其有一个唯一的路径标识;
  • 节点Znode可以包含数据和子节点(但是EPHEMERAL类型的节点不能有子节点);
  • 客户端应用可以在节点上设置监视器。

3.3节点类型

  • Znode有两种类型: 短暂(ephemeral)(断开连接自己删除) 持久(persistent)(断开连接不删除)
  • Znode有四种形式的目录节点(默认是persistent ) PERSISTENT PERSISTENT_SEQUENTIAL(持久序列/test0000000019 ) EPHEMERAL EPHEMERAL_SEQUENTIAL
  • 创建znode时设置顺序标识,znode名称后会附加一个值,顺序号是一个单调递增的计数器,由父节点维护
  • 在分布式系统中,顺序号可以被用于为所有的事件进行全局排序,这样客户端可以通过顺序号推断事件的顺序

使用客户端操作节点

  • 使用命令连接zookeeper服务端:
zkCli.sh -主机名(ip):2181
如:zkCli.sh -mini2:2181
  • 使用 ls 命令来查看当前 ZooKeeper 中所包含的内容:
 ls /
  • 创建一个新的 znode ,使用 create /zk myData 。这个命令创建了一个新的 znode 节点“ zk ”以及与它关联的字符串:
create /zk "myData“
  • 我们运行 get 命令来确认 znode 是否包含我们所创建的字符串:
get /zk

-监听这个节点的变化,当另外一个客户端改变/zk时,输出监听到的变化

get /zk watch
  • 使用set 命令来对 zk 所关联的字符串进行设置:
set /zk "zsl“
  • 使用delete删除 znode 节点:
delete /zk
  • 删除节点(与上面的区别是这个可以删除目录):rmr
rmr /zk

参考文档:http://www.cnblogs.com/likehua/tag/zookeeper/

4.使用java操作zookeeper的api

  • 首先需要引入zookeeper的jar包,这个jar包需要依赖其它的jar,可以直接到maven仓库下载。
    org.apache.zookeeper
    zookeeper
    3.4.10
    pom
  • 常用的增删查改api操作如下: create 在目录树中创建一个节点 delete 删除一个节点 exists 测试是否存在目标节点 get/set data 从目标节点上读取 / 更新数据 get/set ACL 获取 / 设置目标节点访问控制列表信息 get children 检索一个子节点上的列表 sync 等待要被传送的数据 使用java操作代码如下:
public class SimpleZkClient {
    private static final String CONNECT_URL = "mini1:2181,mini2:2181,mini3:2181";

    private static final int SESSION_TIME_OUT = 2000;

    ZooKeeper zkCli = null;
    @Before
    public void init() throws Exception{
        zkCli = new ZooKeeper(CONNECT_URL, SESSION_TIME_OUT, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println(event.getType()+"-----------"+event.getPath());
                try{
                    zkCli.getChildren("/", true);
                }catch (Exception e){

                }

            }
        });
    }

    /**
     * @Description 添加节点数据
     * @Author 刘俊重
     */
    @Test
    public void create() throws Exception{
        // 参数1:要创建的节点的路径 参数2:节点大数据 参数3:节点的权限 参数4:节点的类型。上传的数据可以是任何类型,但都要转成byte[]
        String s = zkCli.create("/zk", "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    /**
     * @Description 判断节点是否存在
     * @Author 刘俊重
     */
    @Test
    public void isExist() throws Exception{
        Stat exists = zkCli.exists("/zk", false);
        System.out.println(null==exists?"不存在":"存在");
    }

    /**
     * @Description 获取节点数据
     * @Author 刘俊重
     */
    @Test
    public void getData() throws Exception{
        byte[] data = zkCli.getData("/zk", false, null);
        System.out.println("节点数据:"+new String(data));
    }

    /**
     * @Description 遍历节点数据
     * @Author 刘俊重
     */
    @Test
    public void getChildren() throws  Exception{
        List children = zkCli.getChildren("/", false);
        for(String s : children){
            System.out.println("节点名称:"+s);
        }
        Thread.sleep(Long.MAX_VALUE);
    }

    /**
     * @Description 删除节点数据
     * @Author 刘俊重
     */
    @Test
    public void delete() throws Exception{
        //参数2:指定要删除的版本,-1表示删除所有版本
        zkCli.delete("/zk",-1);
        this.isExist();
    }

    /**
     * @Description 更新节点数据
     * @Author 刘俊重
     */
    @Test
    public void update() throws Exception{
        Stat stat = zkCli.setData("/zk", "newtest".getBytes(), -1);
        this.getData();
    }
}

Thread.sleep(Long.MAX_VALUE);是为了不让程序执行完之后立马结束,让它睡一会,测试监听是否实现,同时在process回调函数中写了收到通知的操作, zkCli.getChildren("/", true);这时如果我们通过linux命令行操作了zookeeper操作节点就会触发这里的监听事件。

5.zookeeper的使用场景

5.1场景一:客户端动态感知服务端节点变化,实现高可用

现在假设有这样一种需求:服务端节点有多个,可以动态的上下线;需要让任意一台客户端都能实时感知服务端节点的变化,进而连接目前可提供服务的节点。 实现思路:我们可以借助于zookeeper这个第三方中间件,在每台服务器启动时都向zookeeper注册服务器的节点信息(比如:/servers/server01;/servers/server02);客户端每次调用之前都通过getChildren方法获取最新的服务器节点信息,同时客户端在zookeeper注册监听,监听服务器节点的变化;如果某刻服务器server01下线了,zookeeper就会发出节点变化通知客户端,回调process方法拉取最新的服务器节点信息。 服务端代码如下:

public class DistributeServer {

    private static final String CONNECT_URL = "mini1:2181,mini2:2181,mini3:2181";

    private static final int SESSION_TIME_OUT = 2000;

    private static final String PARENT_NODE = "/servers";

    private ZooKeeper zkCli = null;

    /**
     * @Description 获取连接
     * @Author 刘俊重
     * @Date 2017/12/13
     */
    public void getConnect() throws Exception{
        zkCli = new ZooKeeper(CONNECT_URL, SESSION_TIME_OUT, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println(event.getType()+"-----------"+event.getPath());
                try{
                    zkCli.getChildren("/", true);
                }catch (Exception e){

                }

            }
        });
    }

    /**
     * @Description 服务器启动时向zookeeper注册服务信息
     * @Author 刘俊重
     * @Date 2017/12/13
     */
    public void registerServer(String hostName) throws  Exception {
        String s = zkCli.create(PARENT_NODE + "/", hostName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("服务器:"+hostName+"已经注册完毕");
    }

    /**
     * @Description 模拟实际的业务操作
     * @Author 刘俊重
     * @Date 2017/12/13
     */
    public void handelBusiness(String hostname) throws  Exception {
        System.out.println("服务器:"+hostname+"正在处理业务。。。");
        Thread.sleep(Long.MAX_VALUE);
    }

    public static void main(String[] args) throws  Exception {
        DistributeServer server = new DistributeServer();
        server.getConnect();
        server.registerServer(args[0]);
        server.handelBusiness(args[0]);
    }
}

客户端代码如下:

/**
 * @author 刘俊重
 * @Description 模拟客户端,拉取最新服务器节点列表并向zookeeper设置监听
 */
public class DistributeClient {

    private static final String CONNECT_URL = "mini1:2181,mini2:2181,mini3:2181";

    private static final int SESSION_TIME_OUT = 2000;

    private static final String PARENT_NODE = "/servers";

    private ZooKeeper zkCli = null;

    private volatile List serverList = null;

    /**
     * @Description 获取连接
     * @Author 刘俊重
     * @Date 2017/12/13
     */
    public void getConnect() throws Exception{
        zkCli = new ZooKeeper(CONNECT_URL, SESSION_TIME_OUT, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // 收到事件通知后的回调函数(应该是我们自己的事件处理逻辑)
                System.out.println(event.getType()+"-----------"+event.getPath());
                try{
                    //重新更新服务器列表,并且注册了监听
                    getServerList();
                }catch (Exception e){

                }

            }
        });
    }

    /**
     * @Description 获取服务器子节点信息,并对父节点进行监听
     * @Author 刘俊重
     */
    public void getServerList() throws  Exception {
        List children = zkCli.getChildren(PARENT_NODE, true);
        List servers = new ArrayList();
        for(String child : children){
            // child只是子节点的节点名
            byte[] data = zkCli.getData(PARENT_NODE + "/" + child, false, null);
            servers.add(new String(data));
        }
        //把servers赋值给成员变量serverList,以提供给各业务线程使用
        serverList = servers;
        System.out.println("节点数据:"+serverList);

    }
    /**
     * @Description 模拟实际的业务操作
     * @Author 刘俊重
     * @Date 2017/12/13
     */
    public void handelBusiness() throws  Exception {
        System.out.println("客户端开始工作。。。");
        Thread.sleep(Long.MAX_VALUE);
    }

    public static void main(String[] args) throws Exception {
        DistributeClient client = new DistributeClient();

        client.getConnect();
        client.getServerList();
        client.handelBusiness();
    }
}

5.2场景二:分布式锁实现

假设现在集群中有50台机器对某台机器上的同一文件进行修改,如何才能保证这个文件不会被写乱呢,使用java中的synchronized锁肯定是不行的,因为这个锁是对某个程序而言的,而我们这根本就不是在一个服务器上,怎么会锁的住,用zookeeper实现的分布式锁可以实现。 设计思路:服务器启动时都去zookeeper上注册一个“短暂+序号”的znode节点(如/lock/1;/lock/2),并设置监听父节点变化;获取到父节点下所有子节点,并比较序号的大小;约定比如序号最小的获取锁,去操作某一文件,操作完成后删除自己的节点(相当于释放锁),并注册一个新的“短暂+序号”的znode节点;其它程序收到zookeeper发送的节点变化的通知之后,去比较序号的大小,看谁获得新锁。

public class DistributedClientLock {
    // 会话超时
    private static final int SESSION_TIMEOUT = 2000;
    // zookeeper集群地址
    private String hosts = "mini1:2181,mini2:2181,mini3:2181";
    private String groupNode = "locks";
    private String subNode = "sub";
    private boolean haveLock = false;

    private ZooKeeper zk;
    // 记录自己创建的子节点路径
    private volatile String thisPath;

    /**
     * 连接zookeeper
     */
    public void connectZookeeper() throws Exception {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                try {

                    // 判断事件类型,此处只处理子节点变化事件
                    if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {
                        //获取子节点,并对父节点进行监听
                        List childrenNodes = zk.getChildren("/" + groupNode, true);
                        String thisNode = thisPath.substring(("/" + groupNode + "/").length());
                        // 去比较是否自己是最小id
                        Collections.sort(childrenNodes);
                        if (childrenNodes.indexOf(thisNode) == 0) {
                            //访问共享资源处理业务,并且在处理完成之后删除锁
                            doSomething();

                            //重新注册一把新的锁
                            thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
                                    CreateMode.EPHEMERAL_SEQUENTIAL);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        // 1、程序一进来就先注册一把锁到zk上
        thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL_SEQUENTIAL);

        // wait一小会,便于观察
        Thread.sleep(new Random().nextInt(1000));

        // 从zk的锁父目录下,获取所有子节点,并且注册对父节点的监听
        List childrenNodes = zk.getChildren("/" + groupNode, true);

        //如果争抢资源的程序就只有自己,则可以直接去访问共享资源
        if (childrenNodes.size() == 1) {
            doSomething();
            thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);
        }
    }

    /**
     * 处理业务逻辑,并且在最后释放锁
     */
    private void doSomething() throws Exception {
        try {
            System.out.println("gain lock: " + thisPath);
            Thread.sleep(2000);
        } finally {
            System.out.println("finished: " + thisPath);
            //释放锁
            zk.delete(this.thisPath, -1);
        }
    }

    public static void main(String[] args) throws Exception {
        DistributedClientLock dl = new DistributedClientLock();
        dl.connectZookeeper();
        Thread.sleep(Long.MAX_VALUE);
    }

}

参考文档:http://www.cnblogs.com/likehua/tag/zookeeper/

6 zookeeper的选举机制

6.1全新集群paxos

以一个简单的例子来说明整个选举的过程. 假设有五台服务器组成的zookeeper集群,它们的id从1-5,同时它们都是最新启动的,也就是没有历史数据,在存放数据量这一点上,都是一样的.假设这些服务器依序启动,来看看会发生什么.

  1. 服务器1启动,此时只有它一台服务器启动了,它发出去的报没有任何响应,所以它的选举状态一直是LOOKING状态
  2. 服务器2启动,它与最开始启动的服务器1进行通信,互相交换自己的选举结果,由于两者都没有历史数据,所以id值较大的服务器2胜出,但是由于没有达到超过半数以上的服务器都同意选举它(这个例子中的半数以上是3),所以服务器1,2还是继续保持LOOKING状态.
  3. 服务器3启动,根据前面的理论分析,服务器3成为服务器1,2,3中的老大,而与上面不同的是,此时有三台服务器选举了它,所以它成为了这次选举的leader.
  4. 服务器4启动,根据前面的分析,理论上服务器4应该是服务器1,2,3,4中最大的,但是由于前面已经有半数以上的服务器选举了服务器3,所以它只能接收当小弟的命了.
  5. 服务器5启动,同4一样,当小弟.

6.2非全新集群的选举机制(数据恢复)

那么,初始化的时候,是按照上述的说明进行选举的,但是当zookeeper运行了一段时间之后,有机器down掉,重新选举时,选举过程就相对复杂了。 需要加入数据id、leader id和逻辑时钟。 数据id:数据新的id就大,数据每次更新都会更新id。 Leader id:就是我们配置的myid中的值,每个机器一个。 逻辑时钟:这个值从0开始递增,每次选举对应一个值,也就是说: 如果在同一次选举中,那么这个值应该是一致的 ; 逻辑时钟值越大,说明这一次选举leader的进程更新. 选举的标准就变成: 1、逻辑时钟小的选举结果被忽略,重新投票 2、统一逻辑时钟后,数据id大的胜出 3、数据id相同的情况下,leader id大的胜出 根据这个规则选出leader。

您可能感兴趣的

美国分布式商业应用公司CEO黄连金:区块链发展趋势和安全... Catherine 10分钟前 1469 金色财经讯4月28日,由全球区块链精英峰会组委会主办,FansTime承办,并联合国内外多位知名行业学术机构、区块链公司以及顶级专家共同组织的首届“全球区块链精英峰会”今天在中国澳门正式拉开帷幕。...
分布式之数据库和缓存双写一致性方案解析... 引言 为什么写这篇文章? 首先,缓存由于其高并发和高性能的特性,已经在项目中被广泛使用。在读取缓存方面,大家没啥疑问,都是按照下图的流程来进行业务操作。 但是在更新缓存方面,对于更新完数据库,是更新缓存呢,还是删除缓存。又或者是先删除缓存,再更新数据库,其实大家存...
技术干货 | 数据流通关键技术探究 | U刻... 技术干货 | 数据流通关键技术探究 2018年5月23日 栏目:技术分享 数据流通 数据流通是指在数据供方和需方之间按照一定流通规则进行的以数据为对象的行为。数据流通在技术实现上有诸多需求,主要在数据安全、质量保障、权益分配、追溯审计和透...
大家都在说的分布式系统到底是什么?... 在学习分布式之前,先了解一下与之相对应的集中式系统是什么样的。 集中式系统用一句话概括就是:一个主机带多个终端。终端没有数据处理能力,仅负责数据的录入和输出。而运算、存储等全部在主机上进行。现在的银行系统,大部分都是这种集中式的系统,此外,在大型企业、科研单位、军队、政府等也有分布。集中式...
ABP框架 – N层架构 目录 在应用程序设计中,分层架构是一种被广泛使用的技术,它助于降低复杂度和提高代码的可重用性。在ABP框架中,使用了DDD(领域驱动设计)的原则来实现分层架构. DDD分层架构 在DDD(领域驱动设计)架构模型中,有四个基础层。 表现层 : 用户访问接口。使用应用层来实现...
0
稀土掘金

责编内容来自:稀土掘金 (本文源链)
阅读提示:酷辣虫无法对本内容的真实性提供任何保证,请自行验证并承担相关的风险与后果!
本站遵循[CC BY-NC-SA 4.0]。如您有版权、意见投诉等问题,请通过eMail联系我们处理。