四种分布式锁方案

开源拾椹 2021-07-16

1934

“ 分布式锁不是真实的，这听起来可能很滑稽，但并不是说锁本身在分布式系统中是不可能的：只是系统的所有组件都必须参与在指定协议中。在单个程序进程中，线程和线程之间可以使用锁来控制一致性，进程和进程之间则需要分布式锁协议来一致性。”

实现分布式锁的条件

互斥性。在任意时刻，只有一个客户端能持有锁
不会发送死锁。即使一个客户端持有锁的期间崩溃而没有主动释放锁，也需要保证后续其他客户端能够加锁成功
加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给释放了。

—

MySQL分布式锁

分布式锁可以利用mysql的innodb的行锁来实现，有两种方式，悲观锁与乐观锁。悲观锁是

当事务在操作数据时把这部分数据进行锁定，直到操作完毕后再解锁，其他事务操作才可操作该部分数据。这将防止其他进程读取或修改表中的数据；一般实现的方法是select ... for update。乐观锁是如果有人在你之前更新了，你的更新应当是被拒绝的，可以让用户重新操作；一般实现方法给数据加一个版本号，思想跟CAS是一致的，更新前先比较，也就是update video set star = #{star} +1 where star = #{star}.

悲观锁实现方法

package io.jopen.distributelock.mysql;


import org.junit.Before;
import org.junit.Test;


import java.sql.*;
import java.util.concurrent.TimeUnit;


/**
 * 悲观锁（Pessimistic Lock），顾名思义，就是很悲观，每次去拿数据的时候都认为别人会修改，
 * 所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会block直到它拿到锁。
 * <p>
 * 悲观锁：假定会发生并发冲突，屏蔽一切可能违反数据完整性的操作。
 * <p>
 * SQL {select * from users where id =1 for update}
 *
 * @author maxuefeng
 * @since 2019/10/21
 */
public class MySQLPessimisticLockImpl {
    private final String uri = "";
    private Connection conn = null;
    @Before
    public void before() throws ClassNotFoundException, SQLException {
        Class.forName("oracle.jdbc.driver.OracleDriver");
        conn = DriverManager.getConnection(uri, "root", "121101mxf@@ForBigData");
    }


    @Test
    public boolean lock() throws SQLException, InterruptedException {
        conn.setAutoCommit(false);
        String sql;
        while (true) {
            try {
                 悲观锁
                 在查询语句后面增加for update，数据库会在查询过程中给数据库表增加排他锁
                sql = "select * from video where star=1 for update";


                 执行查询
                PreparedStatement ps = conn.prepareStatement(sql);
                ResultSet resultSet = ps.executeQuery();


                return resultSet.next();
            } catch (Exception ignored) {
            }
             线程休眠1秒
            TimeUnit.SECONDS.sleep(1);
        }
    }


    @Test
    public void unlock() throws SQLException {
         释放锁
        conn.commit();
    }
}

乐观锁实现方法（没有unlock，修改完成之后自动释放锁）

package io.jopen.distributelock.mysql;


import org.junit.Before;
import org.junit.Test;


import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;


/**
 * 乐观锁（Optimistic Lock），顾名思义，就是很乐观，每次去拿数据的时候都认为别人不会修改，
 * 所以不会上锁，但是在提交更新的时候会判断一下在此期间别人有没有去更新这个数据。乐观锁适用
 * 于读多写少的应用场景，这样可以提高吞吐量。
 * <p>
 * 乐观锁：假设不会发生并发冲突，只在提交操作时检查是否违反数据完整性。CAS机制
 * 修改之前先进行比较
 * SQL {update video set star = star +1 where `star` = star}
 *
 * @author maxuefeng
 * @since 2019/10/21
 */
public class MySQLOptimisticLockImpl {


     192.168.74.136:3306
    private final String uri = "";
    private Connection conn = null;


    @Before
    public void before() throws ClassNotFoundException, SQLException {
        Class.forName("oracle.jdbc.driver.OracleDriver");
        conn = DriverManager.getConnection(uri, "root", "121101mxf@@ForBigData");
        conn.setAutoCommit(false);
    }


    @Test
    public void lock() throws SQLException, InterruptedException {
        String sql = "update video set star = 2 where star = 1";
        PreparedStatement ps = conn.prepareStatement(sql);
        int updateResult = ps.executeUpdate();


         如果小于0则一直循环 直到获取到锁
        while (updateResult <= 0) {
            updateResult = ps.executeUpdate();
        }
    }
}

笔者用java原生代码实现了基于MySQL数据库的分布式锁的实现，但是编码成本太高，并且MySQL是基于磁盘运行，效率和性能问题相对来说很低，所以笔者不建议各位小伙伴使用MySQL实现分布式锁。

—

Zookeeper分布式锁

zookeeper实现分布式锁跟zookeeper的几个特性有关。

有序节点：假如当前有一个父节点为/lock，我们可以在这个父节点下面创建子节点；zookeeper提供了一个可选的有序特性，例如我们可以创建子节点“/lock/node-”并且指明有序，那么zookeeper在生成子节点时会根据当前的子节点数量自动添加整数序号，也就是说如果是第一个创建的子节点，那么生成的子节点为/lock/node-0000000000，下一个节点则为/lock/node-0000000001，依次类推。

临时节点：客户端可以建立一个临时节点，在会话结束或者会话超时后，zookeeper会自动删除该节点。

事件监听：在读取数据时，我们可以同时对节点设置事件监听，当节点数据或结构变化时，zookeeper会通知客户端。当前zookeeper有如下四种事件：1）节点创建；2）节点删除；3）节点数据修改；4）子节点变更。

实现分布式锁步骤

客户端连接zookeeper，并在/lock下创建临时的且有序的子节点，第一个客户端对应的子节点为/lock/lock-0000000000，第二个为/lock/lock-0000000001，以此类推。
客户端获取/lock下的子节点列表，判断自己创建的子节点是否为当前子节点列表中序号最小的子节点，如果是则认为获得锁，否则监听/lock的子节点变更消息，获得子节点变更通知后重复此步骤直至获得锁；
执行业务代码；
完成业务流程后，删除对应的子节点释放锁。

代码实现（zk客户端选择curator，apache提供的zkClient操作比较繁琐，成本较高）

项目依赖

      <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.0</version>
        </dependency>

客户端配置

    private final String zkQurom = "192.168.74.136:2181";
     ZooKeeper 锁节点路径, 分布式锁的相关操作都是在这个节点上进行
    private final String lockNameSpace = "/distributeLockDir";
    private final RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    private final CuratorFramework zkClient =
            CuratorFrameworkFactory.newClient(
                    zkQurom,
                    5000,
                    3000,
                    retryPolicy);
    @Before
    public void before() {
         启动客户端
        zkClient.start();
    }

加锁（如果获取不到锁，线程会堵塞）

    @Test
    public void curatorForZKDistributeLock() throws Exception {
         创建共享锁
        InterProcessLock lock = new InterProcessSemaphoreMutex(zkClient, lockNameSpace);


         进行加锁
        lock.acquire();


         执行数据修改操作或者说是更新操作


         释放锁
        lock.release();
    }

关闭客户端

    @After
    public void close() {
        CloseableUtils.closeQuietly(zkClient);
    }

—

Redis分布式锁

Redis为单进程单线程模式，Redis本身没有锁的概念，Redis对于多个客户端连接并不存在竞争，但是如果客户端并发访问会出现阻塞链接超时等问题，所以需要锁来解决客户端并发访问的问题，当然这种机制也可以使用到其他业务需求上，redis的shell端的枷锁命令为setnx，setnx [key] [value]，并且设置key的过期时间，不然就会存在死锁的问题。redis目前比较流行的有两种客户端，jedis和redisson，笔者分别用两种客户端都来实现分布式锁。

redisson客户端实现分布式锁

客户端配置

private RedissonClient client = null;
    private final String lockName = "distributeLock";
    @Before
    public void before() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://10.15.41.150:6379").setPassword("12323213");
        client = Redisson.create(config);
    }

加锁（需要注意的是lock这个方法如果获取不到锁的情况下，会一直阻塞线程，直到获取到锁，不抛出线程异常，如果存储锁的服务器崩掉了，会出现死锁的问题，锁释放不了，为了避免这种情况的发生，redisson内置了一个看门狗，他的作用是不断延长锁的有效期）

    @Test
    public void genericDistributeLock1() {
        RLock lock = client.getLock(lockName);
         如果锁不可用，则当前线程变为,出于线程调度目的而禁用，并且处于休眠状态，直到获取锁。
        lock.lock();


         执行数据修改操作或者说是更新操作


         释放锁
        lock.unlock();
    }

redisson公平锁加锁（redisson还提供了一种公平锁，公平锁跟客户端的先来后到有关系，谁先来的谁先获取到锁。）

    **
     * 公平锁 意思是指会按照线程的请求的先来后到进行分配锁的拥有者
     *
     * @see RedissonClient#getFairLock(String)
     * @see RedissonFairLock#lock()
     * @see RedissonFairLock#tryLock()
     */
    @Test
    public void fairLock1() {
         获取锁对象
        RLock fairLock = client.getFairLock(lockName);


         加锁
        fairLock.lock();


         执行数据修改操作或者说是更新操作


         释放锁
        fairLock.unlock();
    }

jedis客户端实现分布式锁

jedis加锁的方式是通过setnx命令进行操作

客户端配置

    private Jedis jedis;
    private final String lockName = "distributeLock";


    @Before
    public void before() {
        jedis = new Jedis("10.55.41.150", 6379);
        jedis.auth("12345");
    }

加锁（px参数是设置过期时间，nx参数是如果没有这个锁，则新建这把锁并且获取，如果加锁成功会返回1，否则为0）

    /**
     * @see SetParams#nx() 不存在时则设置  存在则设置不了值  会返回null
     * @see SetParams#px(long) 设置超时时间  单位是毫秒
     */
    @Test
    public void testGetLock() {
        String lockRes = jedis.set("jedisSetLock", "lockValue", SetParams.setParams().px(5000).nx());
        System.err.println(lockRes);
    }

—

Hazelcast分布式锁

Hazelcast在平时开发中不常见，但是笔者强烈推荐各位读者应该使用Hazelcast作为分布式锁，因为Hazelcast基于内存，并且不依赖中间件，没有master，轻量级运行，只有一个jar。Hazelcast具有各种分布式数据结构，分布式缓存功能，弹性特性，内存缓存支持，与Spring和Hibernate的集成以及更重要的是与众多快乐用户的集成，是功能丰富，面向开发人员友好的内存中数据数据库解决方案。

Hazelcast拓扑图

Hazelcast在其CP Subsystem中引入了接口的线性化分布式实现。它对于粗粒度和细粒度锁定都是有效的。可以使用FencedLock提供的单调防护令牌来实现存在于不同进程中的多个线程之间的互斥。在此博客文章中，您将发现我们如何扩展接口的语义以进行分布式执行，并介绍了分布式设置中可能遇到的几种故障模式。

hazelcast实例配置（声明了两个实例，使用这两个实例来进行竞争锁）

    // 配置对象
    private final Config config = new Config();


    // hazelcastInstance实例
    private HazelcastInstance hazelcastInstance1;
    private HazelcastInstance hazelcastInstance2;


    /**
     * <p>{@link ManagementCenterConfig#ManagementCenterConfig(String, int)}</p>
     *
     * @see Config
     */
    @Before
    public void before() {


        // 创建集群中心配置
        ManagementCenterConfig centerConfig = new ManagementCenterConfig();
        //centerConfig.setUrl("http://127.0.0.1:8200/mancenter");
        centerConfig.setEnabled(true);


        // 设置config
        //config.setInstanceName("test-center")
        config.setManagementCenterConfig(centerConfig)
                .addMapConfig(
                        new MapConfig().setName("mapConfig")
                                .setMaxSizeConfig(new MaxSizeConfig(200, MaxSizeConfig.MaxSizePolicy.FREE_HEAP_SIZE))
                                // 设置删除策略
                                .setEvictionPolicy(EvictionPolicy.LRU)
                                .setTimeToLiveSeconds(20000)
                );
        config.getCPSubsystemConfig().setCPMemberCount(3);


        // 创建hazelcastInstance实例
        hazelcastInstance1 = Hazelcast.newHazelcastInstance(config);
        hazelcastInstance2 = Hazelcast.newHazelcastInstance(config);
    }

加锁

    @Test
    public void HazelcastdistributeLock() {


        // 获取分布式锁
        FencedLock fencedLock1 = hazelcastInstance1.getCPSubsystem().getLock("updateVideoStartLock");
        FencedLock fencedLock2 = hazelcastInstance2.getCPSubsystem().getLock("updateVideoStartLock");


        // 加锁
        boolean tryLock = fencedLock1.tryLock();
        System.err.println(tryLock);


        // 其他安全操作


        // 释放锁
        fencedLock1.unlock();
        boolean tryLock1 = fencedLock2.tryLock();
        System.err.println(tryLock1);
    }