一个简单案例讲解zookeeper实现分布式锁

红太狼 2024-04-17 06:08 67阅读 0赞

整理一下学习的笔记。

用一个例子引出为什么要用分布式锁，假设有一个简单的生成订单id的业务场景，根据时间生成订单序列号，单机运行场景，首先我们可能会这么想。

public class OrderCodeGenerator {
        private int i=0;
    
        public String getOrderCode(){
            Date now = new Date();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd-HH-mm-ss-");
            return sdf.format(now)+ ++i;
        }
    }

接口和接口实现类

//接口类
    public interface OrderService {
        void createOrder();
    }
    
    //实现类
    public class OrderServiceImpl implements OrderService{
        private OrderCodeGenerator ong = new OrderCodeGenerator();
    
        @Override
        public void createOrder() {
            String orderCode= ong.getOrderCode();
            System.out.println(Thread.currentThread().getName()+"=======>"+orderCode);
    
        }
    }
    //测试类
    public class ConcurrentTestDistributeDemo {
        public static void main(String[] args) {
            int currency =50;
            //循环屏障
            CyclicBarrier cb = new CyclicBarrier(currency);
            OrderService orderService = new OrderServiceImpl();
    
            for (int i = 0; i < currency; i++) {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        
                        System.out.println(Thread.currentThread().getName()+"----我准备好了----");
                        try {
                            cb.await();
                        } catch (InterruptedException |BrokenBarrierException e) {
                            e.printStackTrace();
                        }
                        //调用订单服务
                        orderService.createOrder();
                    }
                }).start();
            }
    
    
        }
    
    }

测试类采用50并发数，为了更便于观察，使用juc包的CyclicBarrier的await()，可以理解为50个线程一个个过来被挡住等着，等全部50个到齐了唤醒一起执行，执行后会发现订单号并没有到50，运行一次结果如下（省略前n行），发现其中最后两位数字出现了重复，出现了线程安全问题。

Thread-0----我准备好了----
    Thread-1----我准备好了----
    Thread-2----我准备好了----
    Thread-3----我准备好了----
    //。。省略。。
    Thread-1=======>2019-08-26-15-08-50-39
    Thread-23=======>2019-08-26-15-08-50-48
    Thread-21=======>2019-08-26-15-08-50-49
    Thread-26=======>2019-08-26-15-08-50-47
    Thread-9=======>2019-08-26-15-08-50-46
    
    Process finished with exit code 0

有的小伙伴可能会说直接给生成订单号的方法加同步加锁就可以了，这样不是不可以，但是在高并发的场景下压力都放到业务逻辑处不太好。现在出现了线程安全问题，给实现类加个锁试试，这里要注意锁要是共用的一把锁。

public class OrderServiceImplWithLock implements OrderService{
        private static OrderCodeGenerator ong = new OrderCodeGenerator();
    
    
        private static Lock lock = new ReentrantLock();
    
        @Override
        public void createOrder() {
            String orderCode = null;
            try {
                lock.lock();
                orderCode= ong.getOrderCode();
            }finally {
                lock.unlock();
            }
            System.out.println(Thread.currentThread().getName()+"=======>"+orderCode);
    
        }
    }

问题看起来解决了，如果更大的请求来了呢？多个tomcat部署，分布式的场景，序列号还是唯一的吗？依然可以用代码模拟以下。把demo类new orderservice的语句放到循环run方法中，模拟多个请求同时访问调用各自机器的业务方法。实际这里是会有问题的，因为每个机器都是自己的锁，不是一把锁，分布式场景jvm锁就无法使用了，这时考虑第三方实现。

锁有以下几个特征：

//排他 只有一个线程能获得锁
    //阻塞性 没抢到的阻塞，直到锁释放，再抢
    //可重入性 线程获得锁后，后续是否可重复获得该锁，不然可能会死锁

下面使用zookeeper实现分布式锁，下载，解压即可。我是mac系统，先打开终端，cd到安装目录，bin/zkServer.sh start启动一个服务端。

项目引入zkclient框架，有很多种，选的下面这个。

<dependencies>
            <dependency>
                <groupId>com.101tec</groupId>
                <artifactId>zkclient</artifactId>
                <version>0.10</version>
            </dependency>
        </dependencies>

利用zookeeper的节点唯一性，类似理解为同一个文件夹下只能有一个文件叫这个名字（占有锁），没抢到的可以建立一个监听器观察有锁这个点的状态，一旦删除（释放锁），通知其他节点可以去抢锁。

写一个简单zk的锁实现java lock接口，也可以用自己写的lock接口，代码如下，执行后会发现会打印很多信息，其实每次并不需要通知所有线程，那么如何改进呢？待续。。。。

public class ZKDistributeLock implements Lock {
        private String lockPath;
        private ZkClient client;
    
        public ZKDistributeLock(String lockPath) {
            super();
            this.lockPath = lockPath;
            client = new ZkClient("localhost:2181");
            client.setZkSerializer(new MyZkSerializer());
        }
    
        @Override
        public void lock() {
            if (!tryLock()){
                waitForLock();
                lock();
            }
    
        }
    
        private void waitForLock() {
            //倒计时
            CountDownLatch cdl = new CountDownLatch(1);
            //注册watcher
    
            IZkDataListener listener = new IZkDataListener() {
                @Override
                public void handleDataChange(String s, Object o) throws Exception {
                    System.out.println("收到节点变化");
                }
    
                @Override
                public void handleDataDeleted(String s) throws Exception {
                    System.out.println("收到节点删除");
                   
                    cdl.countDown();
                }
            };
            //注册
    
            client.subscribeDataChanges(lockPath,listener);
            if (this.client.exists(lockPath)){
                try {
                   
    
                    cdl.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //取消注册
            client.unsubscribeDataChanges(lockPath,listener);
        }
    
        @Override
        public void lockInterruptibly() throws InterruptedException {
    
        }
    
        @Override
        public boolean tryLock() {
            try {
                client.createEphemeral(lockPath);
    
            }catch (ZkNodeExistsException e){
                return false;
            }
            return true;
        }
    
        @Override
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
            return false;
        }
    
        @Override
        public void unlock() {
            //删除节点
            client.delete(lockPath);
    
        }
    
        @Override
        public Condition newCondition() {
            return null;
        }
    }