Zookeeper

序号	参数名	说明
7	snapCount	设置多少次事务日志输出后，触发一次快照(snapshot)，此时ZK会生产一个snapshot.*文件，同时创建一个新的事务日志文件log.*,默认100000，真实实现会增加一定的随机数，避免所有服务器同一时间快照影响性能

节点

• create /aaa val创建节点aaa并赋予值val。节点必须有值，否则不能创建

• zk视图结构和标准unix文件系统类似，从/根节点出发。

• 节点成为ZNode,每个节点可存储数据，也可以挂载子节点。因此可以成为树

• 节点类型(不同类型节点名也不能重复)

1. 持久节点，和客户端连接，断开后数据还在（ZNode）
2. 临时节点，和客户端断开后，数据不在
3. 顺序节点，临时节点和持久节点都能创建顺序节点，每次创建节点名都会自动递增(名字+自动生成的序列)

节点状态属性

序号	属性	数据结构	描述
1	czxid	long	节点被创建的Zxid值
2	mzxid	long	节点被修改的Zxid值
3	paxid	long	节点最后一次被修改时的事务ID
4	ctime	long	节点被创建的时间
5	mtime	long	节点最后一次被修改时间
6	dataVersoin	long	节点被修改的版本号(每次修改+1)(CAS保证分布式数据原子性)
7	cversion	long	节点所拥有子节点被修改的版本号
8	aversion	long	节点的ACL被修改的版本号
9	emphemeralOwner	long	如果此阶段为临时节点，这个值就是节点拥有者会话ID,否则0
10	dataLength	long	节点数据域长度
	umChildren	long	节点拥有的子节点个数

ACL

• getAcl /xxx查看xxx的权限

  'world',anyone
  :cdrwa
  

• scheme授权机制, id用户id给谁授权, permissions权限，只读、读写、管理等。

  • create(c)
  • delete(d)
  • read(r)
  • write(w)
  • admin(a) 是否能给子节点设置权限

  机制有

• world，下面只有一个id,叫anyone, world:anyone代表任何人，ZK中对搜有人有权限的节点就是属于world:anyone的

• auth, 它不需要id, 只需要通过authentication的user都有权限。ZK支持通过kerberos来进行authencation，也支持username/password形式的authentication（明文）

• digest, 通过对应的id为username:BASE64(SHA1(password)), 它需要先通过username:password形式的authentication（密文）

• ip, 它对应的id为客户机的IP地址，设置的时候可以设置一个ip段，比如ip:192.168.1.0/16,表示匹配16个bit的IP段

常用ACL命令

• getAcl获取指定节点的ACL信息

• setAcl设置指定节点的ACL信息

  addauth digest username:pwd # 添加用户(会话级别的，退出需要重新操作)
  setAcl /xxx/zzz auth:username:pwd:crwa #给用户添加 auth机制的crwa权限
  setAcl /xxx/zzz digest:username:xxxxmd5:crwa #digest机制的权限添加，密文生产在zookeeper自带的java类 DigestAuthenticationProvider里
  

• addauth注册绘画授权信息

• 注册超级管理员用户可以解决没有权限不能删除的问题

使用

1. zkCli.sh -server ip连接ZooKeeper服务，连接成功后系统会输出相关环境及配置
2. 基本操作
   • 显示根目录下、文件：ls / 查看当前ZooKeeper包含的内容
   • 显示根目录下、文件：ls2 /查看当前节点数据并能看到更新次数等数据
   • 创建文件，并设置初始内容。create /zk "val" 创建一个新的znode节点zk ，以及初始化内容 -e 临时节点(客户端端口就删除) -s 顺序节点(名字自增)
   • 获取内容get /zk 确认znode是否包含我们所创建的字符串
   • 修改set /zk "val2"修改节点内容
   • 删除delete /zk将指定znode删除，如果有子节点，删除失败
   • 递归删除rmr /zk删除节点及子节点
   • 退出quit

四字命令

zk支持使用某些特点四字的命令交互获取服务当前状态，可通过telnet或nc提交命令

1. echo stat|nc ip port查看那个节点被选择作为follower或者leader
2. 使用echo ruok|nc ip port测试是否启动了该server若回复imok表示已经启动
3. echo dump|nc ip port列出未经处理的会话和临时节点
4. echo kill |nc ip port关闭server
5. echo conf | nc ip port输出相关服务配置的详细信息
6. echo cons | nc ip port列出所有连接到服务器的客户端完全的连接/会话的详细信息
7. echo envi|nc ip port输出关于服务环境的详细信息
8. echo reqs|nc ip port列出未经处理的请求
9. echo wchs|nc ip port列出服务器watch的详细信息
10. echo wchc|nc ip port通过session列出服务器的watch详细信息，输出的是一个与watch相关会话的列表
11. echo wchp|nc ip port通过路径列出服务器的watch的详细信息，输出的是与session相关的路径

可视化

1. 事务日志可视化LogFormatter

   java -cp ../../zookeeper-3.4.5.jar;../../lib/slf4j-api-1.6.1.jar org.apache.zookeeper.server.logFormatter log.xxx
   

2. 数据快照可视化SnapshotFormatter

   java -cp ../../zookeeper-3.4.5.jar;../../lib/slf4j-api-1.6.1.jar org.apache.zookeeper.server.SnapshotFormatter snapshot.xxx
   

一致性原理

2pc

2pc 两段提交，强一致性算法。常用于分布式数据库中。

• 术语

1. undo记录原始数据，用于回滚。
2. redo正常提交数据

• 流程

  1. 第一阶段，所有资源数据库都写入undo和redo到事务日志
     2. 第二阶段，所有资源都返回ok，则全部执行commit,否则rollback

• 缺点

  1. 同步阻塞，所有都成功才能成功。
     2. 单点故障，一个返回失败，都失败
     3. 数据不一致，网络延迟导致一个资源commit了，另一个没有commit
     4. 容错机制不完善，一个失败都失败

3pc

先询问资源是否可以访问再进行2pc相同步骤

不一样的是第三阶段协调者如果网络超时或异常，参与者也会commit

优点

1. 改善同步阻塞（不会因为某些访问超时占用时间）
   2. 解决单点故障

paxos算法

少数服从多数，角色轮换避免单点故障

第一阶段，提议者订一个K值，然后访问所有资源（prepare请求），多数回应ok就进行下一阶段，否则k+1再重新请求

第二阶段， 提交数据，绝大部分返回ok则整体成功，否则重新进行第一阶段

问题

1. 主导者故障（单点故障）
2. 最终一致性

协议要求

1. 资源端必须接受第一个prepare
2. 第一个prepare的数据必须要接受

多提议者情况下（解决单点故障）

若干提议者发起prepare,若多数资源同意则进入下一阶段。若同意没超过半数，则k增加再进行prepare。资源会同意更高K的prepare。当提议者认为自己的支持者超过半数就会进行第二阶段，

提交accept,如果资源在次期间遇到更高k的prepare，则会拒绝当前accept，等待最高k发起的提议者的accept。accept接受超过半数则成功，否则k增加，重新prepare。成功后全体接受成功的accept。

ZK使用

Zookeeper原生客户端

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
</dependency>

private final static String CONNECT = "192.168.1.1:8088,xxxx";// 多个用逗号隔开
private static CountDownLatch cdl = new CountDownLatch(1);
ZooKeeper zk = new ZooKeeper(CONNECT, 5000, new Watch(){// 5000超时
    public void process(WatchedEvent watchedEvent) {
        // 如果获取到了连接
        if(watchedEvent.getState() == Event.keeperState.SyncConnected) {
            countDownLatch.countDown();
        }
    }
});

countDownLatch.watch();//等待连接
zookeeper.create("/path", "val".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
Stat stat = new Stat();// 节点状态
// watch为true代表监听此节点，节点内容发生变化会回调连接时注册的watch。watch是一致性的
// watch之后只会返回当前session最后一次修改此节点的内容，即多个setData,只会最后一次回调
byte[] data = zookeeper.getData("/path", true, stat);// 返回值

List<String> childrens = zookeeper.getChildren("/path", true);//[a,b,c]的形式返回子节点

ACL acl = new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvicer.generateDigest("root:root")));
ACL acl2 = new ACL(ZooDefs.Perms.CREATE, new Id("ip", "192.168.1.1"));
List<ACL> acls = new ArrayList<>();
acls.add(acl);
acls.add(acl2);
// 创建持久节点
zookeeper.create("path", "val".getBytes(), acls, CreateMode.PERSISTENT);
// 添加digest方案的权限
zookeeper.addAuthInfo("digest", "root:root".getBytes());

问题

1. 会话连接是异步的
2. watch需要重复注册，一次watch只能监听一个
3. 缺少session重连机制
4. 复杂，缺少很多功能，例如级联新增

ZkClient

<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
</dependency>

private final static String CONNECT_HOST = "192.168.1.1:8000";
// zkClient不用监听
ZkClient zkClient = new ZkClient(CONNECT_HOST, 4000);
// 提供了递归创建父节点的功能,true代表级联新增，false代表不允许（父节点不存在就会报错）
zkClient.createPersistent("/zk/zkclient/zkclient1", true);
// 获取子节点 [node1,node2,node3]
List<String> list = zkClient.getChildren("/zkclient");
// 监听
// 节点内容修改
zkClient.subscribeDataChange("/nodeName", new IZkDataListener() {
    public void handleDataChange(String nodeName, Object newVal) throws Exception {
        // nodeName节点名称  newVal修改后的值
    }
    
    public void handleDataDeleted(String nodeName) throws Exception {
        // ...
    }
});

// nodeName中的子节点发升变化触发
zkClient.subscribeChildChanges("/nodeName", new IZkChildListener()) {
    public void handleChildChange(String nodeName, List<String> list) throws Exception {
        // nodeName节点名 list节点列表
    }
}
// 监听器
// subscriptStateChanges
// 权限
public void addAuthInfo(String scheme, final byte[] auth);
public void setAcl(final String path, final List<ACL> acl);

Curator

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
</dependency>

private final static String CONNECT_HOST="192.168.1.1:8000";
// 尝试三次连接，失败1000毫秒后重试，第二次重试时间2*1000毫秒，第三次3*1000
CuratorFramework curatorFramework = CuratorFrameworkFactory.
    newCliebt(CONNECT_HOST, 5000, 5000, new ExponentialBackoffRetry(1000, 3));
curatorFramework.start();// 启动连接

// 另一种写法
CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().connectString(CONNECT_HOST).sessionTimeoutMs(5000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
curatorFramework.start();// 启动连接
// 创建节点,返回节点路径
String val = curatorFramework.create()
    .createingParentsIfNeeded()//级联创建父节点
    .withMode(CreateMode.PERSISTENT)//持久节点
    .forPath("/path/path1/path2", "val".getBytes());
// 删除节点,级联删除
curatorFramework.delete().deleteChildrenIfNeed().forPath("/path");

// 异步执行
final CountDownLatch countDownLatch = new CountDownLatch(1);
ExcutorService service = Executors.newFixedThreadPool(1);
curatorFramework.create().createingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
    .inBackground(new BackgroundCallback(){
        public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throw Exception {
            curatorFramework.getResultCode();// 结果
            curatorFramework.getType();
            countDownLatch.countDown();
        }
    },server).forPath("/path/path1/path2", "val".getBytes());    
// 事务, 同时成功才成功
Collection<CuratorTransactionResult> resultCollections = curatorFramework.inTransaction()
    .create().forPath("/path/path1/path2", "val".getBytes()).and()
    .setData().forPath("/path/path1/path2", "v1".getBytes()).and().commit();

for(CuratorTransactionResult res: resultCollections) {
    res.getForPath();// 节点路径
    res.getType();// 结果
}

// watch机制
// Pathcache 监听一个路径下子节点的创建、删除、数据更新
// NodeCache 节点的创建、删除、更新
// TreeCache Pathcache+NodeCache
NodeCache nodeCache = new NodeCache(curatorFramework, "/curator", false);
nodeCache.start(true);
nodeCache.getListenable().addListener(()->System.out.println("节点变化，变成了" + new String(nodeCache.getCurrentData().getData())));

pathCache.getListenable().addListener((framework, event)->{
    event.getType();//CHILD_ADD CHILD_REMOVED CHILD_UPDATE 子节点增加、删除、更新
});

集群

特点

1. 顺序一致性，命令执行顺序一致
2. 原子性，集群中所有机器都成功，否则失败
3. 单一视图，连接集群任意一个机器数据一致
4. 可靠性，一个更新被操作之前，数据不变
5. 实时性，一个节点更改，其他节点很短时间内同步
6. 角色轮换，避免故障

角色

1. leader任务调度，事务处理(增，删，改)
2. follower非事务请求，读。参与投票
3. observer观察者，读，不参与投票。（3.30以上提供，增加效率）

配置集群

zoo.cfg

server.0=192.168.1.2:2333:2444 # 选举端口：通讯端口
server.1=192.168.1.3:2333:2444 # 1代表 myid ，集群名称，必须是数字
server.2=192.168.1.4:2333:2444

ZAB协议

类似paxos,zk自己实现的协议