Zookeeper--08---zk实现分布式锁案例

Posted 高高for 循环

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Zookeeper--08---zk实现分布式锁案例相关的知识,希望对你有一定的参考价值。

提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档


zk实现分布式锁

1.zk中锁的种类:

  • 读锁:⼤家都可以读,要想上读锁的前提:之前的锁没有写锁
  • 写锁:只有得到写锁的才能写。要想上写锁的前提是,之前没有任何锁。

2.zk如何上读锁

1.创建⼀个临时序号节点,节点的数据是read,表示是读锁

2.获取当前zk中序号⽐⾃⼰⼩的所有节点

3.判断最⼩节点是否是读锁

  • 如果不是读锁的话,则上锁失败,为最⼩节点设置监听。阻塞等待,zk的watch机制
    会当最⼩节点发⽣变化时通知当前节点,于是再执⾏第⼆步的流程
  • 如果是读锁的话,则上锁成功

3.zk如何上写锁

1.创建⼀个临时序号节点,节点的数据是write,表示是 写锁

2.获取zk中所有的⼦节点

3.判断⾃⼰是否是最⼩的节点

  • 如果是,则上写锁成功
  • 如果不是,说明前⾯还有锁,则上锁失败,监听最⼩的节点,如果最⼩节点有变化, 则回到第⼆步。

4.⽺群效应

如果⽤上述的上锁⽅式,只要有节点发⽣变化,就会触发其他节点的监听事件,这样的话对
zk的压⼒⾮常⼤,——⽺群效应。

可以调整成链式监听。解决这个问题。

5.curator实现读写锁

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;


@SpringBootTest
public class TestReadWriteLock 

    @Autowired
    private CuratorFramework client;

    @Test
    void testGetReadLock() throws Exception 
        // 读写锁
        InterProcessReadWriteLock interProcessReadWriteLock=new InterProcessReadWriteLock(client, "/lock1");
        // 获取读锁对象
        InterProcessLock interProcessLock=interProcessReadWriteLock.readLock();
        System.out.println("等待获取读锁对象!");
        // 获取锁
        interProcessLock.acquire();
        for (int i = 1; i <= 100; i++) 
            Thread.sleep(3000);
            System.out.println(i);
        
        // 释放锁
        interProcessLock.release();
        System.out.println("等待释放锁!");

    

    @Test
    void testGetWriteLock() throws Exception 

        // 读写锁
        InterProcessReadWriteLock interProcessReadWriteLock=new InterProcessReadWriteLock(client, "/lock1");
        // 获取写锁对象
        InterProcessLock interProcessLock=interProcessReadWriteLock.writeLock();
        System.out.println("等待获取写锁对象!");
        // 获取锁
        interProcessLock.acquire();
        for (int i = 1; i <= 100; i++) 
            Thread.sleep(3000);
            System.out.println(i);
        
        // 释放锁
        interProcessLock.release();
        System.out.println("等待释放锁!");
    



分布式锁案例

案例分析

  • 比如说 "进程 1"在使用该资源的时候,会先去获得锁在使用该资源的时候,保持独占,这样其他进程就无法访问该资源,
  • "进程1"用完该资源以后就将锁释放掉,让其他进程来获得锁,
  • 那么通过这个锁机制,我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

  1. 接收到请求后,在/locks节点下创建一个临时顺序节点
  2. 判断自己是不是当前节点下最小的节点:是,获取到锁;不是,对前一个节点进行监听
  3. 获取到锁,处理完业务后,delete节点释放锁,然后下面的节点将收到通知,重复第二步判断

依赖:

   <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.5.7</version>
        </dependency>

分布式锁实现

  • 获取连接
  • 对zk加锁
  • 对zk解锁

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock 

    private final String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private final int sessionTimeout = 2000;
    private final ZooKeeper zk;

    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);

    private String waitPath;
    private String currentMode;

    public DistributedLock() throws IOException, InterruptedException, KeeperException 

        // 获取连接
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() 
            @Override
            public void process(WatchedEvent watchedEvent) 
                // connectLatch  如果连接上zk  可以释放
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected)
                    connectLatch.countDown();
                

                // waitLatch  需要释放
                if (watchedEvent.getType()== Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath))
                    waitLatch.countDown();
                
            
        );

        // 等待zk正常连接后,往下走程序
        connectLatch.await();

        // 判断根节点/locks是否存在
        Stat stat = zk.exists("/locks", false);

        if (stat == null) 
            // 创建一下根节点
            zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        
    

    // 对zk加锁
    public void zklock() 
        // 创建对应的临时带序号节点
        try 
            currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // wait一小会, 让结果更清晰一些
            Thread.sleep(10);

            // 判断创建的节点是否是最小的序号节点,如果是获取到锁;如果不是,监听他序号前一个节点

            List<String> children = zk.getChildren("/locks", false);

            // 如果children 只有一个值,那就直接获取锁; 如果有多个节点,需要判断,谁最小
            if (children.size() == 1) 
                return;
             else 
                Collections.sort(children);

                // 获取节点名称 seq-00000000
                String thisNode = currentMode.substring("/locks/".length());
                // 通过seq-00000000获取该节点在children集合的位置
                int index = children.indexOf(thisNode);

                // 判断
                if (index == -1) 
                    System.out.println("数据异常");
                 else if (index == 0) 
                    // 就一个节点,可以获取锁了
                    return;
                 else 
                    // 需要监听  他前一个节点变化
                    waitPath = "/locks/" + children.get(index - 1);
                    zk.getData(waitPath,true,new Stat());

                    // 等待监听
                    waitLatch.await();

                    return;
                
            


         catch (KeeperException e) 
            e.printStackTrace();
         catch (InterruptedException e) 
            e.printStackTrace();
        


    

    // 解锁
    public void unZkLock() 

        // 删除节点
        try 
            zk.delete(this.currentMode,-1);
         catch (InterruptedException e) 
            e.printStackTrace();
         catch (KeeperException e) 
            e.printStackTrace();
        

    



测试


import org.apache.zookeeper.KeeperException;

import java.io.IOException;

public class DistributedLockTest 

    public static void main(String[] args) throws InterruptedException, IOException, KeeperException 

       final  DistributedLock lock1 = new DistributedLock();

        final  DistributedLock lock2 = new DistributedLock();

       new Thread(new Runnable() 
           @Override
           public void run() 
               try 
                   lock1.zklock();
                   System.out.println("线程1 启动,获取到锁");
                   Thread.sleep(5 * 1000);

                   lock1.unZkLock();
                   System.out.println("线程1 释放锁");
                catch (InterruptedException e) 
                   e.printStackTrace();
               
           
       ).start();

        new Thread(new Runnable() 
            @Override
            public void run() 

                try 
                    lock2.zklock();
                    System.out.println("线程2 启动,获取到锁");
                    Thread.sleep(5 * 1000);

                    lock2.unZkLock();
                    System.out.println("线程2 释放锁");
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            
        ).start();

    


对比单体模式下—ReentrantLock

Curator框架实现分布式锁案例

依赖

       <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-client</artifactId>
            <version>4.3.0</version>
        </dependency>

获取客户端连接

  private static CuratorFramework getCuratorFramework() 

        ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);

        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("hadoop102:2181,hadoop103:2181,hadoop104:2181")
                .connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(policy).build();

        // 启动客户端
        client.start();

        System.out.println("zookeeper 启动成功");
        return client;
    

测试案例


import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorLockTest 
    public static void main(String[] args) 

        // 创建分布式锁1
        InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");

        // 创建分布式锁2
        InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");

        new Thread(new Runnable() 
            @Override
            public void run() 
                try 
                    lock1.acquire();
                    System.out.println("线程1 获取到锁");

                    lock1.acquire();
                    System.out.println("线程1 再次获取到锁");

                    Thread.sleep(5 * 1000);

                    lock1.release();
                    System.out.println("线程1 释放锁");

                    lock1.release();
                    System.out.println("线程1  再次释放锁");

                 catch (Exception e) 
                    e.printStackTrace();
                
            
        ).start();

        new Thread(new Runnable() 
            @Override
            public void run() 
                try 
                    lock2.acquire();
                    System.out.println("线程2 获取到锁");

                    lock2.acquire();
                    System.out.println("线程2 再次获取到锁");

                    Thread.sleep(5 * 1000);

                    lock2.release以上是关于Zookeeper--08---zk实现分布式锁案例的主要内容,如果未能解决你的问题,请参考以下文章

Redis实现分布式锁与Zookeeper实现分布式锁区别

二Redis实现分布式锁

基于redis(lua)和zookeeper分布式锁(秒杀)实现,分布式接口幂等实现,分布式速率限制实现,分布式ID生成器实现

Redis实现分布式锁

使用Redis实现分布式锁

用 MySQL 实现分布式锁,你听过吗?