17.Zookeeper

Zookeeper分布式锁实现

1、单应用场景下的锁机制

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4yNVBYQ3-1595351027916)(C:\Users\CJ-xi\AppData\Roaming\Typora\typora-user-images\image-20200709023142491.png)]

在这种情况下如何处理呢？

1、数据库乐观锁：

数据更新状态发生变化后就停止更新，使用version记录的方式

select * from im_product_inventory where variants_id="v1001"; /**version=10**/
update im_product_inventory set inventory=inventory-1,version=version+1 where variants_id="v1001" and version=10;

优点：防止脏数据写入

缺点：没有顺序概念的，没有等待机制，共同抢夺资源，版本异常并不代表数据就无法实现，并且并发会对数据库带来压力

2、悲观锁

假设数据肯定会冲突，使用悲观锁一定要关闭自动自动提交set autocommit=0;

/**开始事务**/
begin transaction;
select inventory from product_inventory where id=1 for update;
/**开始修改**/
update product_inventory set inventory=inventory-1 where id=1;
/**提交事务**/
commit;

一定要注意：select … for update，MySQL InneDB默认是行锁，行级锁都是基于索引实现的，如果查询条件不是索引列，会把整个表锁组，原因是如果没有索引就会扫整个表，就会锁表

优点：单任务，事务的隔离原子性

缺点：数据库效率执行低下，容易导致系统超时锁表

随着互联网系统的三高架构：高并发、高性能、高可用的提出，悲观锁用的越来越少

3、程序方式

synchronized(this){}

并发等待，并且只有一个线程能进入

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-avDG6Yq4-1595351027919)(C:\Users\CJ-xi\AppData\Roaming\Typora\typora-user-images\image-20200709023517632.png)]

2、分布式场景下的锁机制

乐观锁悲观锁解决方案

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ClFaidhn-1595351027920)(C:\Users\CJ-xi\AppData\Roaming\Typora\typora-user-images\image-20200709023536851.png)]

1、分布式锁应该具备哪些条件

• 在分布式系统环境下，一个方法在同一时间只能被一个机器一个线程执行
• 高可用的获取锁和释放锁
• 高性能的获取与释放锁
• 具备可重入特性（多个线程同时进入也要保证不出错误）
• 具备锁失效机制，防止死锁
• 具备非阻塞特性，即便没有获取到锁也要能返回锁失效

2、分布式锁有哪些实现方式

• 使用Redis内存数据来实现分布式锁
• zookeeper来实现分布式锁
• Chubby：Google，Paxos算法解决

3、通过Redis实现分布式锁理解基本概念

分布式锁有三个核心要素：加锁、解锁、锁超时

• 加锁：操作资源先去Redis里查是否有锁这个资源
  • get？setnx命令，key可以用资源的唯一标识，比如库存id，value？
  • 如果这个key存在就返回给用户：目前资源紧张请等待。或者循环访问直到可以使用为止
• 解锁：操作完成后通过del释放锁，这个时候另一个线程就能重新获取锁
• 锁超时：如果JVM1在执行业务过程中宕机了，这个时候锁就驻留Redis中无法释放，这个时候就会死锁，JVM2就永远无法执行了，这个时候就需要通过expire 10超时来设置锁的生命周期了

以上三步只是简单的实现了分布式锁的逻辑，上面三个操作有几个致命问题

1、非原子性操作

• setnx ，JVM1宕机了，expire
• 需要在加锁的时候就把超时时间一并设置成功

2、误删锁

自己的锁因为超时而删除，这个时候下一个线程创建了一个新锁，新锁会被上一个线程锁删除，怎么办

• 锁是谁的吧？value存UUID，线程ID
• 删除的时候判断这个锁里的UUID或线程ID是不是自己，是的话才删除

3、执行的先后顺序

我的执行线程以外还需要一个守护线程，锁超时时间是10秒，每隔9秒或更短来检测一下执行线程是否完成，如果没有完成就延长锁的失效时间，给他续命

4.Redis实现分布式锁机制

分布式锁的核心

• 加锁
• 锁删除：一定不能因为锁提前超时导致删除非自己的锁
• 锁超时：如果业务没有执行完就应该给锁延时

<dependency>
  <groupId>org.redisson</groupId>
  <artifactId>redisson-spring-boot-starter</artifactId>
  <version>3.12.1</version>
</dependency>

Java实现代码

@Configuration
public class controller {

    @Autowired
    RedissonClient redissonClient;

    @GetMapping("/redisson")
    @ResponseBody
    public String redissonLock(){
        RLock rLock = redissonClient.getLock("orderId");
        System.out.println("开启锁**********");
        try {
            //如果有锁等待5秒，加锁后持有30秒
            rLock.tryLock(5, 30, TimeUnit.SECONDS);
            System.out.println("获取锁*********");
        }catch (Exception ex){
            ex.printStackTrace();
        } finally {
            System.out.println("释放锁*********");
            rLock.unlock();
        }
        return "redisson";
    }
}

redisson默认就是加锁30秒，建议也是30秒以上，默认的lockWatchdogTimeout会每隔10秒观察一下，待到20秒的时候如果主进程还没有释放锁，就会主动续期30秒

3、zookeeper内部结构

zookeeper是一种分布式协调的服务，通过简单的API解决我们的分布式协调问题

对于zookeeper来讲，其实就是一个树形的目录存储结构，但和目录不同的是节点本身也可以存放数据，每个节点就是一个znode

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-O10icIPf-1595351027922)(C:\Users\CJ-xi\AppData\Roaming\Typora\typora-user-images\image-20200709024257185.png)]

znode由以下四个部分组成

• data：znode存放数据的地方（一个znode节点能存放的数据大小被限制在1MB以内）
• ACL：访问权限
• stat：各种元数据
• child：当前节点的子节点引用

#bin目录下启动 ./zkServer.sh start
#bin目录下启动客户端 ./zkCli.sh 
ls / #列出目录信息
create /node_1 Java #创建节点及节点值
delete /node_1  #删除节点
rmr /node_1 #删除节点及子节点
set /node_1 Jsp #设置节点值

znode有这两个特性：一个是顺序（非顺序），一个是临时（持久）组合出了四种节点方式

• 顺序持久节点
• 非顺序持久节点
• 顺序临时节点
• 非顺序临时节点

Zookeeper的事件通知

就是zk自带的Watch机制，可以理解为注册在特定znode上的触发器，只要这个znode发生改变（删除，修改）会触发znode上注册的对应事件，请求获取这个节点Watch的节点就会收到异步的通知

4、zookeeper实现分布式锁的原理

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qTWmfNpb-1595351027924)(C:\Users\CJ-xi\AppData\Roaming\Typora\typora-user-images\image-20200709024314124.png)]

• 加锁：创建临时非顺序节点
• 解锁：删除锁
• 锁失效：临时节点就是锁失效

5、zookeeper业务代码实现分布式锁

接口实现

public interface ZookeeperLock {
    public void lock();
    public void unlock();
}

抽象类

public abstract class AbstractZookeeperLock implements ZookeeperLock {

    protected String lock = "/";
    protected String zk_address = "127.0.0.1:2181";
    protected ZkClient zkClient = new ZkClient(zk_address);
    protected CountDownLatch countDownLatch;

    @Override
    public void lock() {
        //尝试获取锁
        if(tryLock()){
            //拿到锁
            System.out.println("获得锁成功........");
        }else{
            //获得锁失败，等待获取锁，阻塞
            waitLock();
            //等待结束获取锁
            lock();
        }
    }

    @Override
    public void unlock() {
        //创建的是临时节点
        //关闭连接来解锁
        if(zkClient!=null){
            //关闭连接就可以删除节点
            zkClient.close();
            System.out.println("关闭连接释放锁......");
        }
    }

    protected abstract boolean tryLock();
    protected abstract boolean waitLock();
}

实现类

public class ZookeeperDistributeLock extends AbstractZookeeperLock {

    //初始化的时候就传递lock路径
    public ZookeeperDistributeLock(String lock_path){
        lock = lock_path;
    }

    @Override
    protected boolean tryLock() {
        try {
            //创建一个临时节点
            zkClient.createEphemeral(lock);
            System.out.println("创建节点成功.........");
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("创建节点失败.........");
            return false;
        }
    }

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

            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                if(countDownLatch!=null){
                    //倒计数器
                    countDownLatch.countDown();
                }
            }
        };
        //订阅一个数据改变的通知
        zkClient.subscribeDataChanges(lock,iZkDataListener);
        if(zkClient.exists(lock)){
            countDownLatch = new CountDownLatch(1);
            try {
                countDownLatch.await();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        //取消订阅
        zkClient.unsubscribeDataChanges(lock,iZkDataListener);
        return false;
    }
}

new CountDownLatch(1);
try {
countDownLatch.await();
}catch (Exception e){
e.printStackTrace();
}
}
//取消订阅
zkClient.unsubscribeDataChanges(lock,iZkDataListener);
return false;
}
}

本文地址：https://blog.csdn.net/weixin_42031645/article/details/107502562