Zookeeper实现分布式锁

Posted 920913cheng

tags:

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

在单节点上运行的程序,多个线程对一个共享变量进行操作,则有可能出现线程安全的问题。例如:春运期间购买火车票

public class BuyTicketRunnable implements Runnable 
    //static 表示ticketCount是全局共享变量
    public static int ticketCount = 100;

    @Override
    public void run() 
        while (true) 
            if (ticketCount > 0) 
                try 
                    // 此处休眠500毫秒更能看出效果
                    Thread.sleep(20);
                    ticketCount--;
                    System.out.println("当前剩余票的数量:" + ticketCount);
                 catch (InterruptedException e) 
                    e.printStackTrace();
                    return;
                
            

        
    

    public static void main(String[] args) 
        for (int i = 0; i < 10; i++) 
            new Thread(new BuyTicketRunnable()).start();
        
    

当前剩余票的数量:15
当前剩余票的数量:11
当前剩余票的数量:9
当前剩余票的数量:9
当前剩余票的数量:9
当前剩余票的数量:8
当前剩余票的数量:5
当前剩余票的数量:3
当前剩余票的数量:4
当前剩余票的数量:6
当前剩余票的数量:6
当前剩余票的数量:2
当前剩余票的数量:0
当前剩余票的数量:1
当前剩余票的数量:-1
当前剩余票的数量:-2
当前剩余票的数量:-3
当前剩余票的数量:-5
当前剩余票的数量:-7
当前剩余票的数量:-5
当前剩余票的数量:-6

最后运行的结果显示和实际情况有出入的,之所以出现不是连续的减少车票,是因为多个线程是并行执行的,从结果看出了多个线程操作共享数据,会出现线程安全问题,所以我们为了确保这种问题不会出现,可以使用锁.

锁的分类有很多:读写锁,悲观锁,乐观锁,重入锁等...........

Synchronized

synchronized属于重入锁,

重入锁,也叫做递归锁,指的是同一线程 外层函数获得锁之后 ,内层递归函数仍然有获取该锁的代码,但不受影响。

 

 

public class BuyTicketRunnable implements Runnable 
    // static 表示ticketCount是全局共享变量
    public static int ticketCount = 100;
    //lock作为全局变量锁
    public static Object lock = new Object();

    @Override
    public void run() 
        synchronized (BuyTicketRunnable.lock) 
            while (true) 
                if (ticketCount > 0) 
                    try 
                        // 此处休眠500毫秒更能看出效果
                        Thread.sleep(20);
                        ticketCount--;
                        System.out.println("当前剩余票的数量:" + ticketCount);
                     catch (InterruptedException e) 
                        e.printStackTrace();
                        return;
                    
                

            
        
    

    public static void main(String[] args) 
        for (int i = 0; i < 10; i++) 
            new Thread(new BuyTicketRunnable()).start();
        
    

 

当前剩余票的数量:29
当前剩余票的数量:28
当前剩余票的数量:27
当前剩余票的数量:26
当前剩余票的数量:25
当前剩余票的数量:24
当前剩余票的数量:23
当前剩余票的数量:22
当前剩余票的数量:21
当前剩余票的数量:20
当前剩余票的数量:19
当前剩余票的数量:18
当前剩余票的数量:17
当前剩余票的数量:16
当前剩余票的数量:15
当前剩余票的数量:14
当前剩余票的数量:13
当前剩余票的数量:12
当前剩余票的数量:11
当前剩余票的数量:10
当前剩余票的数量:9
当前剩余票的数量:8
当前剩余票的数量:7
当前剩余票的数量:6
当前剩余票的数量:5
当前剩余票的数量:4
当前剩余票的数量:3
当前剩余票的数量:2
当前剩余票的数量:1
当前剩余票的数量:0

 

使用lock锁

lock锁需要手动加锁,手动释放锁,如果遇到异常,锁还在不会自动释放,Synchronized锁如果遇到异常,则会自动释放锁

 

在分布式或者集群中可能出现的数据安全问题

 技术图片

 Zookeeper实现分布式锁原理

    zookeeper是以树形结构的方式存储数据,每个节点不能重复,zookeeper里面有四种节点,两大类节点,持久节点和临时节点。持久节点是存放到本地里,连接断了也存在,但是临时节点如果zk的连接断掉以后。临时节点则不会存在,正是因为这种特性,我们可以使用zk的临时节点来作为锁。

 有A,B两个连接操作数据,A连接先去创建节点,如果能创建节点成功,表示获取了锁可以操作数据,此时如果B连接在来创建节点,根据zk存储特性,节点名字不能有重复,此时B连接是无法创建节点成功,所以处于等待状态。等待A连接处理完数据以后,断开连接,临时节点断开连接以后该节点就消失(释放锁),此时B 就可以在创建节点(获取锁),B处理完以后断开连接(释放锁)。所以zk作为锁的原理是,临时节点创建成功(获取锁),断开连接(释放锁).

 

JAVA代码实现Zookeeper分布式锁

  导入Zookeeper所需要的依赖

        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.14</version>
        </dependency>
        <dependency>
            <groupId>com.101tec</groupId>
            <artifactId>zkclient</artifactId>
            <version>0.10</version>
        </dependency>

 增接口:

package com.zk.distributed;

public interface DistributedLock 
    // 获取锁方法
    void getLock();

    // 释放锁方法
    void unLock();

抽象类实现接口:

package com.zk.distributed;

import java.util.concurrent.CountDownLatch;

import org.I0Itec.zkclient.ZkClient;

public abstract class ZkDistributedAbstractLock implements DistributedLock 
    // zk连接地址
    private  final String CONNETION_ADDR = "127.0.0.1:2181";
    // 创建zk连接
    protected ZkClient zkClient = new ZkClient(CONNETION_ADDR);
    //创建这个节点,以这个节点作为锁
    protected  final String PATH = "/Distributed";
    //信号量
    protected CountDownLatch countDownLatch = null;

    @Override
    public void getLock() 
        if (tryGetLock()) 
            System.out.println("###(创建节点成功)获取锁成功#####");
         else 
            // 等待
            waitUnLock();
            // 重新获取锁
            getLock();
        
    

    // 尝试去获取锁
    abstract Boolean tryGetLock();

    // 如果当前锁被占用,则等待
    abstract void waitUnLock();

    @Override
    public void unLock() 
        if (zkClient != null) 
            zkClient.close();
            System.out.println("关闭连接,释放资源(释放锁)");
        
    

实现抽象类:

package com.zk.distributed;

import java.util.concurrent.CountDownLatch;

import org.I0Itec.zkclient.IZkDataListener;

public class ZkDistributeLock extends ZkDistributedAbstractLock 

    @Override
    Boolean tryGetLock() 
        try 
            zkClient.createEphemeral(PATH);
            return true;
         catch (Exception e) 
            return false;
        

    

    @Override
    void waitUnLock() 

        // 使用事件监听,获取到节点被删除,zookeeper上节点有什么变化都可以感知到
        IZkDataListener iZkDataListener = new IZkDataListener() 
            // 当节点被删除
            public void handleDataDeleted(String dataPath) throws Exception 
                if (countDownLatch != null) 
                    // 唤醒
                    countDownLatch.countDown();
                

            

            // 当节点发生改变
            public void handleDataChange(String dataPath, Object data) throws Exception 

            
            
        ;

        // 注册节点信息
        zkClient.subscribeDataChanges(PATH, iZkDataListener);
        //检测是否存在该节点信息
        if (zkClient.exists(PATH)) 
            // 创建信号量
            countDownLatch = new CountDownLatch(1);
            try 
                // 等待
                countDownLatch.await();
             catch (Exception e) 

            

        
        // 删除事件通知
        zkClient.unsubscribeDataChanges(PATH, iZkDataListener);
    

测试类:

package com.zk.distributed;

import com.zk.distributed.DistributedLock;
import com.zk.distributed.ZkDistributeLock;

public class DistributedRunnable implements Runnable 
    public DistributedLock lock = new ZkDistributeLock();
    public static int count = 0;

    @Override

    public void run() 
        try 
            // 上锁
            lock.getLock();
            Thread.sleep(1000);
            // 模拟用户生成订单号
            count++;
            System.out.println("当前累加########" + count);
            // getNumber();
         catch (Exception e) 
            e.printStackTrace();
         finally 
            // 釋放鎖資源
            lock.unLock();
        
    

    public static void main(String[] args) 
        for (int i = 0; i < 25; i++) 
            new Thread(new DistributedRunnable()).start();
        
    

 

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

Zookeeper分布式锁

zooKeeper实现分布式锁

基于zookeeper的分布式锁实现

zookeeper怎么实现分布式锁

基于Zookeeper实现的分布式互斥锁 - InterProcessMutex

ZooKeeper实现读写锁