rpc系列-ZooKeeper

public class DistributedServer {
    private static final String host = "localhost:2181";
    private static final int sessionTimeout = 2000;
    private static final String parentNode = "/servers/";

    private ZooKeeper zk = null;

    /**
     * 创建到zk的客户端连接
     *
     * @throws Exception
     */
    public void getConnect() throws Exception {

        zk = new ZooKeeper(host, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // 收到事件通知后的回调函数
                System.out.println(event.getType() + "__" + event.getPath());
                try {
                    zk.getChildren("/", true);
                } catch (Exception e) {
                }
            }
        });

    }

    /**
     * 向zk集群注册服务器信息
     * ZooDefs.Ids.OPEN_ACL_UNSAFE   默认匿名权限，权限scheme id：'world,'anyone，权限位：31（adcwr）
     * ZooDefs.Ids.READ_ACL_UNSAFE   只读权限，权限scheme id：'world,'anyone，权限位：1（r）
     *
     * CreateMode
     * 节点类型，类型定义在枚举CreateMode中：
     * （1）PERSISTENT：持久；
     * （2）PERSISTENT_SEQUENTIAL：持久顺序；
     * （3）EPHEMERAL：临时；
     * （4）EPHEMERAL_SEQUENTIAL：临时顺序。
     * @param data  创建节点初始化内容
     * @throws Exception
     */
    public void registerServer(String data) throws Exception {
        String create = zk.create(parentNode + "test", data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(data + " 注册节点 " + create);
    }

    /**
     * 业务功能
     *
     * @throws InterruptedException
     */
    public void handleBussiness(String data) throws InterruptedException {
        System.out.println(data + "开始handleBussiness");
        Thread.sleep(Long.MAX_VALUE);
    }

    public static void main(String[] args) throws Exception {

        // 获取zk连接
        DistributedServer server = new DistributedServer();
        server.getConnect();

        // 利用zk连接注册服务器信息
        server.registerServer("test01");

        // 启动业务功能
        server.handleBussiness("test01");

    }

}

/**
 *
 *分布式锁：几种实现方式，示例用临时顺序节点实现共享锁的一般做法
 *
 * 逻辑
 * 1.zk上注册一个"临时+序号"的znode,并监听父节点
 * 2.获取父节点下所有程序子节点，比较序号大小
 * 3.序号最小的获取到"锁"，去访问资源，访问完后，删除自己的节点，释放锁，重新注册一个新的子节点
 * 4.其他程序节点会收到事件通知，可以去zk上获取锁
 */
public class DistributedClientLock {

    // 会话超时
    private static final int SESSION_TIMEOUT = 2000;
    // zookeeper集群地址
    private String hosts = "localhost:2181";
    private String groupNode = "servers";
    private String subNode = "test";
    private boolean haveLock = false;

    private ZooKeeper zk;
    /**
     * 记录自己创建的子节点路径
     * volatile 不是线程安全的，具有可见性，在一个子内存操作完后，立即刷新回到主内存。
     * 如果不加Volatile，每次调用thisPath，会有副本，修改会有延迟，比如其它线程抢到没有修改完的数据，就在新的线程继续执行，造成最后数据值有误
     * 比如：一个线程写，其它线程去读的时候，用的Volatile，比如监听新节点插入。
     */
    private volatile String thisPath;

    /**
     * 连接zookeeper
     */
    public void connectZookeeper() throws Exception {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {
            public void process(WatchedEvent event) {
                try {
                    System.out.println(event.getType()+"____"+event.getPath());

                    /**
                     * 判断事件类型，此处只处理子节点变化事件
                     *                          event For “/path”                           event For “/path/child”
                     * create(“/path”)          EventType.NodeCreated                       无
                     * delete(“/path”)          EventType.NodeDeleted       	            无
                     * setData(“/path”)         EventType.NodeDataChanged                   无
                     * create(“/path/child”)    EventType.NodeChildrenChanged（getChild)     EventType.NodeCreated
                     * delete(“/path/child”)    EventType.NodeChildrenChanged（getChild)     EventType.NodeDeleted
                     * setData(“/path/child”)   无                                           EventType.NodeDataChanged
                     */
                    if (event.getType() == Event.EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {
                        //获取子节点，并对父节点进行监听
                        List<String> 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, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                                    CreateMode.EPHEMERAL_SEQUENTIAL);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

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

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

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

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

    /**
     * 处理业务逻辑，并且在最后释放锁
     */
    private void doSomething() throws Exception {
        try {
            System.out.println("锁: " + thisPath);
            Thread.sleep(2000);

        } finally {
            System.out.println("完成: " + 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);
    }
}