zookeeper分布式锁

核心：

1, 基于zookeeper创建 EPHEMERAL （临时节点），谁抢到谁就能创建订单号，没抢到就等待它释放（事件通知），再去抢。

2, 抢到锁，生成订单号，调用close()方法 （释放临时节点）。

标红的都是zookeeper的特性，用来实现分布式锁的关键。

分布式锁实现顺序：

package com.hyw.zklook;

public interface ZkLock {
    
    void getLock();

    
    void waitLock();

    
    void unLock();

    
    boolean tryLock();
}

package com.hyw.zklook;


import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.CreateMode;

import java.util.concurrent.CountDownLatch;

public class ZkLockImp implements ZkLock {
    //参数1 zk地址
    static final String ADDRESS = "127.0.0.1:2181";
    //参数2 zk超时时间
    static final int TIME_OUT = 5000;
    //临时节点
    final String path = "/lockPath";
    //计数器
    CountDownLatch countDownLatch = null;
    //zk连接
    ZkClient zkClient = new ZkClient(ADDRESS, TIME_OUT);
    @Override
    public void getLock() {
        if(tryLock()){  //获取锁成功

        }else{
            //等待
            waitLock();
            //重新获取锁，递归
            getLock();
        }
    }

    
    @Override
    public boolean tryLock() {
        try {
            //创建锁成功
            zkClient.create(path, path, CreateMode.EPHEMERAL);
            System.out.println(Thread.currentThread().getName()+"获取锁成功！");
            return true;
        } catch (RuntimeException e) {
//            e.printStackTrace();
        }
        return false;
    }

    
    @Override
    public void waitLock() {
        IZkDataListener zkDataListener = new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {

            }
            //监听节点是否被删除
            @Override
            public void handleDataDeleted(String s) throws Exception {
//                System.out.println("锁被释放了，可以抢锁了。。。");
                if(countDownLatch != null){
                    countDownLatch.countDown();
                }
            }
        };
        //注册当前监听事件通知
        zkClient.subscribeDataChanges(path, zkDataListener);
        if (zkClient.exists(path)){
            if(countDownLatch == null){
                countDownLatch = new CountDownLatch(1);
            }
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //移除监听事件通知
        zkClient.unsubscribeDataChanges(path, zkDataListener);
    }

    @Override
    public void unLock() {
        if(null != zkClient){
            zkClient.close();
            System.out.println(Thread.currentThread().getName()+"释放锁成功！");
        }
    }

}

3，创建订单service类OrderService

package com.hyw.zklook;

public class OrderService implements Runnable {
    private ZkLock lock = new ZkLockImp();

    @Override
    public void run() {
        getNumber();
    }

    private void getNumber(){
        try {
            lock.getLock();
            String number = generatorNumber();
            System.out.println(Thread.currentThread()+",获取的number:"+number);
        } finally {
            lock.unLock();
        }
    }

    static Integer number = 50000000;   //初始化单号

    
    static String generatorNumber(){
        String dh = "单号："+number;
        number ++;

        return dh;
    }
}

package com.hyw.zklook;

public class TestMain {
    public static void main(String[] args) {
        for (int i = 0; i < 1000; i++) {
            new Thread(new OrderService()).start();
        }
    }
}

效果：