zookeeper 分布式锁的实现及原理

保证线程安全

ZooKeeper初始化示意图

InterProcessMutex

public static void main(String[] args) {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
                // 客户端连接任意zk 节点都行,也可以指定集群
                .connectString(CommonConstants.CONNECT_URL)
                .sessionTimeoutMs(5000)
                // 重试策略
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .connectionTimeoutMs(4000).build();
        // 表示启动
        curatorFramework.start();
        /**
         * locks表示命名空间
         *
         * 锁的获取逻辑是放在zookeeper上
         * 当前锁是跨进程可见
         */
        InterProcessMutex lock = new InterProcessMutex(curatorFramework, "/locks");
        for (int i = 0; i < 10; i++) {
            new Thread(()->{
                System.out.println(Thread.currentThread().getName() + "->" + "尝试抢占锁");
                try {
                    // 抢占锁，没有抢占到，则阻塞
                    lock.acquire();
                    System.out.println(Thread.currentThread().getName() + "->" + "获取锁成功");
                } catch (Exception e) {
                    e.printStackTrace();
                }
                // 释放锁
                try {
                    Thread.sleep(4000);
                    lock.release();
                    System.out.println(Thread.currentThread().getName() + "->" + "释放锁成功");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            },"t-"+i).start();
        }

    }

zookeeper leader 选举

kafka -> master/slave | kafka+zookeeper

Leader latch

　　利用临时有序节点

Leader selector

利用 -> curator.recepis 中锁的实现

• dubbo + zookeeper 注册中心，注册服务

• dubbo + zookeeper 配置中心，元数据管理

• 实现分布式锁（Curator）

• leader 选举（定时任务的互斥执行）

  • leader latch

• leader selector(释放leader权限知乎，仍然可以继续参与选举)

zookeeper实际应用以及原理分析

　　Zookeeper的基本原理

数据模型

• 弱一致性模型

• 2 PC协议（原子性）

• 过半提交

ZABZAB(zookeeper atomic Broadcast)

 

• leader选举

• 数据同步

• 崩溃恢复 （looking）

• 原子广播

•  

消息广播

• Zxid(64)

• observer（不参与投票和ack,只和leader保持数据 同步）

崩溃恢复

• 选举出新leader（选举谁作为leader）

• 数据同步

Zxid(*)

• 已经被处理的消息不能丢失

• 被丢弃的消息不能再次出现

如何实现？（leader）

 

• zxid最大（和leader选举有关系？）

• Epoch ->term(raft).zxid()

Zxid(事务id)

• 64位

• 高32位是epoch、低32位代表递增是不编号