zoukankan      html  css  js  c++  java
  • 分布式协调Zookeeper(分布式锁&Leader选举)

    分布式协调-Zookeeper(分布式锁&Leader选举)

    在微服务的情况下,我们通常会通过集群部署去缓解节点压力,而如果有多个用户同时去抢一个商品,如果我们后端不去做处理,那肯定就出现问题。而传统的synchronized是无法解决跨进程的问题的。那我们肯定就要引入一个第三方的视角去帮我们来解决这个问题,那zk的一些特性就能帮助我们去实现分布式锁的问题。

    • 【节点的唯一性】:因为zk上的节点名称是唯一的,那我们就可以多个客户端同时创建同样名称的节点在zk上,创建成功的节点就是获取到锁的节点,而没有创建成功的节点,就可以通过zk的watcher机制,去监听节点的删除事件,一旦节点被删除,其他的节点就可以去创建文件。
      • 问题可能会产生惊群效应,这对于zkserver的通信是一负担。
    • 【有序节点的特性】:所有的客户端都去创建一个临时有序节点并且他们都是在一个容器节点下(一定时间内,容器节点下没有子节点,容器节点就会被删除),那么最小的节点就是获得锁的客户端,没有获得锁的就相等于正在排队。他们要做的就是监听他们下一个节点的删除事件,这样就解决了惊群效应问题。

    curator已经提供了一些这样的实现,我们下面写一个demo,并且去分析它的源码实现。

     curator实现分布式锁

    pom

    <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>5.2.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>5.2.0</version> </dependency>
    View Code

    注入curator

    @Configuration public class CuratorConfig { @Bean public CuratorFramework curatorFramework(){ CuratorFramework curatorFramework=CuratorFrameworkFactory .builder() .connectString("192.168.221.128:2181") .sessionTimeoutMs(15000) .connectionTimeoutMs(20000) .retryPolicy(new ExponentialBackoffRetry(1000,10)) .build(); curatorFramework.start(); return curatorFramework; } }
    View Code

    创建多个线程去模拟客户端去扣减库存,使用curator的提供的有序节点的api去实现锁功能的完整性。

     @GetMapping("{goodsNo}")
        public String  purchase(@PathVariable("goodsNo")Integer goodsNo) throws Exception {
            QueryWrapper<GoodsStock> queryWrapper=new QueryWrapper<>();
            queryWrapper.eq("goods_no",goodsNo);
            //基于临时有序节点来实现的分布式锁.
            InterProcessMutex lock=new InterProcessMutex(curatorFramework,"/Locks");
            try {
                //抢占分布式锁资源(阻塞的)
                lock.acquire();
                GoodsStock goodsStock = goodsStockService.getOne(queryWrapper);
                Thread.sleep(new Random().nextInt(1000));
                if (goodsStock == null) {
                    return "指定商品不存在";
                }
                if (goodsStock.getStock() < 1) {
                    return "库存不够";
                }
                goodsStock.setStock(goodsStock.getStock() - 1);
                boolean res = goodsStockService.updateById(goodsStock);
                if (res) {
                    return "抢购书籍:" + goodsNo + "成功";
                }
            }finally {
                lock.release(); //释放锁
            }
            return "抢购失败";
        }
    View Code

    Curator实现分布式锁的源码分析

    流程为:

    获取锁:判断自己是否只最小的节点,如果是则获取锁,不是则循环获取锁,对上一个节点进行一次性监听使用wait进行等待,当收到监听事件的时候,则对线程使用notify进行唤醒。

    删除锁:因为synchronize是重入锁,所以首先需删除锁的重入次数,然后删除存储在map中的节点即可,随后删除zk中的节点即可。

    获取锁

    private boolean internalLock(long time, TimeUnit unit) throws Exception
    {
      
        
        Thread currentThread = Thread.currentThread();
        //判断当前线程是否已经获取到锁
        LockData lockData = threadData.get(currentThread);
        
        // 当前线程已经获取过锁(锁的重入性质)
        if ( lockData != null )
        {
            // re-entering
            lockData.lockCount.incrementAndGet();
            return true;
        }
    
        // 尝试获得锁
        String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
        if ( lockPath != null )
        {
            //构建一个锁的数据
            LockData newLockData = new LockData(currentThread, lockPath);
            //保存在map中
            threadData.put(currentThread, newLockData);
            return true;
        }
    
        return false;
    }
    String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
    {
        final long      startMillis = System.currentTimeMillis();
        final Long      millisToWait = (unit != null) ? unit.toMillis(time) : null;
        final byte[]    localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
        int             retryCount = 0;
    
        String          ourPath = null;
        boolean         hasTheLock = false;
        boolean         isDone = false;
        while ( !isDone )
        {
            isDone = true;
    
            try
            {
                //创建一个临时有序节点,并且返回当且节点的名称
                ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
                //通过循环的操作去获得锁
                hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
            }
            catch ( KeeperException.NoNodeException e )
            {
                if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
                {
                    isDone = false;
                }
                else
                {
                    throw e;
                }
            }
        }
    
        //如果已经获得了锁,则返回
        if ( hasTheLock )
        {
            return ourPath;
        }
    
        return null;
    }
    private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
    {
        boolean     haveTheLock = false;
        boolean     doDelete = false;
        try
        {
            if ( revocable.get() != null )
            {
                client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
            }
    
           //只要没有获得锁,并且连接是启动状态,就一直循环,不断开
            while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
            {
                //获得排序后的list
                List<String>        children = getSortedChildren();
                //获取创建节点的序列号
                String      sequenceNodeName = ourPath.substring(basePath.length() + 1); 
                //比较序列号是否是最小的,如果是最小的则返回,否则返回比自己小1的节点,并且包装成PredicateResults
                PredicateResults    predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
                // 获得锁修改haveTheLock为true则退出循环
                if ( predicateResults.getsTheLock() )
                {
                    haveTheLock = true;
                }
                else
                {
                    //当前节点的上一个节点
                    String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
    
                    synchronized(this)
                    {
                        try
                        {
                        //针对上一个节点进行一次性监听  
                        client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                            if ( millisToWait != null )
                            {
                                millisToWait -= (System.currentTimeMillis() - startMillis);
                                startMillis = System.currentTimeMillis();
                                if ( millisToWait <= 0 )
                                {
                                    doDelete = true;    // timed out - delete our node
                                    break;
                                }
    
                                wait(millisToWait);
                            }
                            else
                            {
                                wait();
                            }
                        }
                        catch ( KeeperException.NoNodeException e )
                        {
                            // it has been deleted (i.e. lock released). Try to acquire again
                        }
                    }
                }
            }
        }
        catch ( Exception e )
        {
            ThreadUtils.checkInterrupted(e);
            doDelete = true;
            throw e;
        }
        finally
        {
            if ( doDelete )
            {
                deleteOurPath(ourPath);
            }
        }
        return haveTheLock;
    }

    【删除锁】:

    @Override
    public void release() throws Exception
    {
        /*
            Note on concurrency: a given lockData instance
            can be only acted on by a single thread so locking isn't necessary
         */
    
        Thread currentThread = Thread.currentThread();
        LockData lockData = threadData.get(currentThread);
        // 无法从映射表中获取锁信息,表示当前没有持有锁
        if ( lockData == null )
        {
            throw new IllegalMonitorStateException("You do not own the lock: " + basePath);
        }
        // 锁是可重入的,初始值为1,原子-1到0,锁才释放
        int newLockCount = lockData.lockCount.decrementAndGet();
        if ( newLockCount > 0 )
        {
            return;
        }
        if ( newLockCount < 0 )
        {
            throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath);
        }
        try
        {
            // lockData != null && newLockCount == 0,释放锁资源
            internals.releaseLock(lockData.lockPath);
        }
        finally
        {
            // 最后从映射表中移除当前线程的锁信息
            threadData.remove(currentThread);
        }
    }
    final void releaseLock(String lockPath) throws Exception
    {
        //移除订阅事件
        client.removeWatchers();
        revocable.set(null);
        // 删除临时顺序节点,只会触发后一顺序节点去获取锁,理论上不存在竞争,只排队,非抢占,公平 锁,先到先得
        deleteOurPath(lockPath);
    }

      curator实现leader选举

    实际上底层还是使用zk的有序节点的特性,谁小谁就是leader。我们这里使用quartz去写一个demo。当一个quartz挂了的时候,别的quartz马上执行。实际上就是,使用wather监听别的节点,一旦leader释放了锁(自己是最小的),那就成为leader.

    使用一个工厂bean,其中维护一个状态(是否是leader),当spring启动的时候去判断这个状态,如果是leader,则执行定时任务。

    //SchedulerFactoryBean是一个工程bean,通过这个把quartz的信息转移到Spring中,可以对定时任务进行触发
    public class ZkSchedulerFactoryBean extends SchedulerFactoryBean {
    
        private LeaderLatch leaderLatch;
    
        //namespace
        private final String LEADER_PATH="/leader";
    
    
        public ZkSchedulerFactoryBean() throws Exception {
            //在Spring启动的时候不自动开启定时任务
            this.setAutoStartup(false);
            leaderLatch=new LeaderLatch(getClient(),LEADER_PATH);
            //当leader发生变化的时候,会回调LeaderLatchListener
            leaderLatch.addListener(new GlenLeaderLatchListener(this));
            //开始监听
            leaderLatch.start();
        }
    
        //连接zk
        private CuratorFramework getClient(){
            CuratorFramework curatorFramework= CuratorFrameworkFactory
                    .builder()
                    .connectString("192.168.43.3:2181")
                    .sessionTimeoutMs(15000)
                    .connectionTimeoutMs(20000)
                    .retryPolicy(new ExponentialBackoffRetry(1000,10))
                    .build();
            curatorFramework.start();
            return curatorFramework;
        }
    
        //创建一个定时调度的实例
        @Override
        protected void startScheduler(Scheduler scheduler, int startupDelay) throws SchedulerException {
            //如果是启动状态的话,当当前节点抢到leader的时候,就会把这个状态设置为true,然后定时任务就启动了
            if(this.isAutoStartup()) {
                super.startScheduler(scheduler, startupDelay);
            }
        }
    
        //释放资源
        @Override
        public void destroy() throws SchedulerException {
            CloseableUtils.closeQuietly(leaderLatch);
            super.destroy();
        }
    }

    zk的监听类,如果自己是leader,则对上面的工厂类中的状态进行修改,定时任务启动

    public class GlenLeaderLatchListener implements LeaderLatchListener {
        //控制定时任务启动和停止的方法
        private SchedulerFactoryBean schedulerFactoryBean;
    
        GlenLeaderLatchListener(SchedulerFactoryBean schedulerFactoryBean) {
            this.schedulerFactoryBean = schedulerFactoryBean;
        }
    
        //当他是leader的时候,把他的状态设置启动状态
        @Override
        public void isLeader() {
            System.out.println(Thread.currentThread().getName()+"成为了leader");
            schedulerFactoryBean.setAutoStartup(true);
            schedulerFactoryBean.start();
        }
    
        //如果他不是leader,就停止执行定时任务
        @Override
        public void notLeader() {
            System.out.println(Thread.currentThread().getName()+"抢占leader失败,不执行任务");
            schedulerFactoryBean.setAutoStartup(false);
            schedulerFactoryBean.stop();
        }
    }

    执行定时任务的类

    public class QuartzJob extends QuartzJobBean {
    
        @Override
        protected void executeInternal(JobExecutionContext jobExecutionContext) {
            System.out.println("开始执行定时任务");
            SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.println("当前执行的系统时间:"+sdf.format(new Date()));
        }
    }

    把这些东西交给Spring

    @Configuration
    public class QuartzConfiguration {
    
    
        @Bean
        public ZkSchedulerFactoryBean schedulerFactoryBean(JobDetail jobDetail,Trigger trigger) throws Exception {
            ZkSchedulerFactoryBean zkSchedulerFactoryBean=new ZkSchedulerFactoryBean();
            zkSchedulerFactoryBean.setJobDetails(jobDetail);
            zkSchedulerFactoryBean.setTriggers(trigger);
            return zkSchedulerFactoryBean;
        }
    
        //执行定时任务的类
        @Bean
        public JobDetail jobDetail(){
            return JobBuilder.newJob(QuartzJob.class).storeDurably().build();
        }
    
        //触发定时任务的类,1s执行一次,永远执行
        @Bean
        public Trigger trigger(JobDetail jobDetail){
            SimpleScheduleBuilder simpleScheduleBuilder=
                    SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(1).repeatForever();
            return TriggerBuilder.newTrigger().forJob(jobDetail).withSchedule(simpleScheduleBuilder).build();
        }
    }

    我们启动两个SpingBoot应用

     

     发现zk上有一个leader节点,并且有一个客户端已经抢占到leader

    这个时候停止其中一个节点,另外一个节点马上就监测到了leader的变化,然后开始执行

     因为他已经是最小节点

     

  • 相关阅读:
    Asp.NET 4.0 ajax实例DataView 模板编程1
    ASP.NET 4.0 Ajax 实例DataView模板编程 DEMO 下载
    部分东北话、北京话
    .NET 培训课程解析(一)
    ASP.NET 4.0 Ajax 实例DataView模板编程2
    ASP.NET Web Game 架构设计1服务器基本结构
    ASP.NET Web Game 构架设计2数据库设计
    TFS2008 基本安装
    Linux上Oracle 11g安装步骤图解
    plsql developer远程连接oracle数据库
  • 原文地址:https://www.cnblogs.com/UpGx/p/15599521.html
Copyright © 2011-2022 走看看