[DB] Zookeeper

介绍

• 相当于“数据库”，类似linux、hdfs的属性文件结构
• 分布式协调框架，实现HA（High Availability）
• 分布式锁管理框架
• 保证数据在zookeeper集群之间数据的事务一致性

组成

• Leader：数据总控节点，用于接收客户端连接请求，分发给所有的Follower节点后，各个Follower节点进行更新数据操作并返回给Leader节点，如果满足半数以上（所以Zookeeper集群一般是奇数个节点）更新成功则此次操作成功
• Follower：跟随者，Zookeeper的Leader宕机时，所有的Follower内部通过选举机制，选举出新的Leader
• Observer：客户端，用于观察Zookeeper集群的数据发送变化，如果产生变化则zookeeper会主动推送watch事件给Observer（客户端），用于对数据变化的后续处理；Observer（客户端）也可以发送数据变更请求

功能

• 分布式协调
• 分布式锁
• 数据同步（端口：2888）
• 选举机制（端口：3888）

场景

• 统一命名服务
• 配置管理
• 集群管理
  • Hadoop2.0使用ZK的事件处理确保整个集群只有一个活跃的NameNode，并存储配置信息
  • HBase使用ZK的事件处理确保整个集群只有一个HMaster，察觉HRegionServer联机和宕机，并存储访问控制列表
• 共享锁
• 队列管理

单机配置

• 解压，设置环境变量
• 生效环境变量
• 创建目录 /root/training/zookeeper-3.4.10/tmp
• 在 conf/ 下创建zoo_sample.cfg的副本zoo.cfg，添加内容：
• dataDir=/root/training/zookeeper-3.4.10/tmp
• server.1=bigdata111:2888:3888
• 在 /root/training/zookeeper-3.4.10/tmp 下创建文件：myid，输入1

•  zkCli.sh：连接客户端
  • create /node1 "hello world"

秒杀系统

•  实现方式
  • Zookeeper：利用分布式锁
  • Redis：利用队列
  • 消息系统：利用队列

传统多线程方式（会产生混乱）

package demo;

public class TestZookeeperDemo {
    
    
    private static int NUMBER = 10;
    private static void getNumber() {
        System.out.println("*******业务方法开始***********");
        System.out.println("当前值：" + NUMBER);
        NUMBER --;
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("*******业务方法结束***********");
    }
    
    public static void main(String[] args) {
        //启动10个线程
        for(int i = 0 ; i < 10 ; i ++ ) {
            new Thread(new Runnable() {
                public void run() {
                    getNumber();
                }
            }).start();
        }
    }
}

Zookeeper方式（分布式锁）

package demo;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class TestZookeeperDemo {
    
    
    private static int NUMBER = 10;
    private static void getNumber() {
        System.out.println("*******业务方法开始***********");
        System.out.println("当前值：" + NUMBER);
        NUMBER --;
        
        // 睡两秒，客户端依然拿着锁
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("*******业务方法结束***********");
    }
    
    public static void main(String[] args) {
        // 定义retry policy（重试策略）
        // 每次等待1000ms，重试10次
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        // 创建一个ZK的客户端
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                                .connectString("192.168.174.111:2181")
                                .retryPolicy(policy)
                                .build();
        
        cf.start();
        
        //在ZK中定义一把锁
        final InterProcessMutex lock = new InterProcessMutex(cf, "/mylock");
        
        //启动10个线程
        for(int i = 0 ; i < 10 ; i ++ ) {
            new Thread(new Runnable() {
                public void run() {
                    try {
                    // 请求得到锁，如果没有得到，使用retrypolicy重试
                    lock.acquire();
                    
                    // 访问共享资源
                    getNumber();
                    }catch(Exception ex) {
                        ex.printStackTrace();
                    }finally {
                        // 释放锁
                        try {
                            lock.release();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
        }
    }
}

  ...

ZK集群

• Leader（bagdata113）
• Follower（bigdata112、bigdata114）