分布式技术（一）分布式锁

秒速五厘米 2022-06-15 12:07 202阅读 0赞

**1. 数据库锁**  
有两种方式：  
（1）利用唯一键（主键）：数据库是有唯一主键规则的，主键不能重复，对于重复的主键会抛出主键冲突异常。当我们想要锁住某个方法时，执行以下SQL：

insert into methodLock(method_name,desc) values (‘method_name’,‘desc’);

因为我们对method\_name做了唯一性约束，这里如果有多个请求同时提交到数据库的话，数据库会保证只有一个操作可以成功，那么我们就可以认为操作成功的那个线程获得了该方法的锁，可以执行方法体内容。当方法执行完毕之后，想要释放锁的话，需要执行以下SQL:

delete from methodLock where method_name ='method_name';

**优缺点**：

*  严重依赖数据库的可用性。
 *  没有超时机制，一旦超时操作失败，锁将无法释放。
 *  没有阻塞机制，一旦获取锁失败，就会失败返回，需要重新发起获取锁的请求再次尝试获取锁。
 *  无法重入，同一个锁的持有者在没有释放锁的前提下无法重新获取锁。

（2）利用排它锁：在查询语句后面增加for update，数据库会在查询过程中给数据库表增加排他锁。当某条记录被加上排他锁之后，其他线程无法再在该行记录上增加排他锁。我们可以认为获得排它锁的线程即可获得分布式锁，当获取到锁之后，可以执行方法的业务

public void unlock(){
         connection.commit();//释放锁
     }

**优缺点：**  
\- 严重依赖数据库的可用性。  
\- 相比第一种方式，在一定程度上解决了超时的问题，服务宕机超过一定时间数据库服务器会自动断掉，从而释放锁。  
\- 可以在数据库服务器端设置for update的机制为等待而不是立即失败返回，可视为一种阻塞机制（如果阻塞请求较多会占用大量数据库连接）。  
\- 仍然无法可重入。  
**2. Redis缓存锁**  
setnx()命令，setnx的含义就是SET if Not Exists，其主要有两个参数 setnx(key, value)。该方法是原子的，如果key不存在，则设置当前key成功，返回1；如果当前key已经存在，则设置当前key失败，返回0。但是要注意的是setnx命令不能设置key的超时时间，只能通过expire()来对key设置。

public boolean lock(String key, long timeout) {
                      boolean lockSuccess = false;
                      try {
                             long start = System.currentTimeMillis();
                             String lockKey = GlobalIdInitializer.DEFAULT_PREFIX_OF_LOCK + key;
                             do {
                                 long result = setnx(lockKey, String.valueOf(
                                 System.currentTimeMillis() + GlobalIdInitializer.LOCKKEY_EXPIRE_TIME + 1));
                                 if (result == 1) {
                                     lockSuccess = true;
                                     break;
                                 } else {
                                     String lockTimeStr = getStringFromRedis(lockKey);
                                     if (StringUtils.isNumeric(lockTimeStr)) {
       // 如果key存在，锁存在
                                     long lockTime = Long.valueOf(lockTimeStr);
                                     if (lockTime < System.currentTimeMillis()) {
       // 锁已过期
                                          String originStr = getSet(lockKey,
                                          String.valueOf(System.currentTimeMillis()
                                                + GlobalIdInitializer.LOCKKEY_EXPIRE_TIME + 1));
                                          if (StringUtils.isNoneBlank(originStr)
                                               && originStr.equals(lockTimeStr)) {
       // 表明锁由该线程获得
                                               lockSuccess = true;
                                               break;
                                          }
                                    }
                                 }
                             }
                             // 如果不等待，则直接返回
                             if (timeout == 0) {
                                 break;
                             }
                             // 等待300ms继续加锁
                             Thread.sleep(300);
                          } while ((System.currentTimeMillis() - start) < timeout);
    
                    } catch (Exception e) {
                       logger.error(e.getMessage());
                    } finally {
                  }
                  return lockSuccess;
                }

public void unLock(String key) {
                      try {
                             String lockTimeStr = getStringFromRedis(key);
                             long lockTime = Long.valueOf(lockTimeStr);
                             /* * 判断锁是否过期，如果锁过期，说明锁可能被其他进程获得，这时直接delete * 会把其他进程已获得的锁释放掉 */
                             if (lockTime > System.currentTimeMillis()) {
                                  deleteKey(GlobalIdInitializer.DEFAULT_PREFIX_OF_LOCK + key);
                             }
                       } catch (Exception e) {
                           logger.error(e.getMessage());
                       } finally {
                       }
               }

**优缺点：**  
\- 有单点问题，可能出现数据丢失。  
**3. RedLock**  
redis作者鉴于单点redis作为分布式锁的可能出现的锁数据丢失问题，提出了Redlock算法，该算法实现了比单一节点更安全、可靠的分布式锁管理（DLM）。算法的步骤如下：  
（1）客户端获取当前时间，以毫秒为单位。  
（2）客户端尝试获取N个节点的锁，（每个节点获取锁的方式和前面说的缓存锁一样），N个节点以相同的key和value获取锁。客户端需要设置接口访问超时，接口超时时间需要远远小于锁超时时间，比如锁自动释放的时间是10s，那么接口超时大概设置5-50ms。这样可以在有redis节点宕机后，访问该节点时能尽快超时，而减小锁的正常使用。  
（3）客户端计算在获得锁的时候花费了多少时间，方法是用当前时间减去在步骤一获取的时间，只有客户端获得了超过3个节点的锁，而且获取锁的时间小于锁的超时时间，客户端才获得了分布式锁。  
（4）客户端获取的锁的时间为设置的锁超时时间减去步骤三计算出的获取锁花费时间。  
（5）如果客户端获取锁失败了，客户端会依次删除所有的锁。使用Redlock算法，可以保证在挂掉最多2个节点的时候，分布式锁服务仍然能工作，这相比之前的数据库锁和缓存锁大大提高了可用性，由于redis的高效性能，分布式缓存锁性能并不比数据库锁差。  
**优缺点：**  
\- Redis所有节点之间是独立的，必须由客户端控制写入的一致性。  
\- 当集群中若干节点宕机时，客户端需要等到超时时间之后才会返回，影响性能 。  
\- 会有冲突造成假死锁问题。如果5个节点，由于需要获取3个节点以上的锁才算成功获取锁，如果都获取了1-2个节点的锁，那么没有一个客户端能够成功获取锁。redis作者借鉴了raft算法的精髓，通过冲突后在随机时间开始，可以大大降低冲突时间，但是这问题并不能很好的避免，特别是在第一次获取锁的时候，所以获取锁的时间成本增加了。两个Redlock的问题，最关键的一点在于Redlock需要客户端去保证写入的一致性，后端5个节点完全独立，所有的客户端都得操作这5个节点。如果  
**4. Zookeeper分布式锁**  
提到分布式协调服务，自然就想到了zookeeper。zookeeper实现了类似paxos协议，是一个拥有多个节点分布式协调服务。对zookeeper写入请求会转发到leader，leader写入完成，并同步到其他节点，直到所有节点都写入完成，才返回客户端写入成功。zookeeper还有几个特质，让它非常适合作为分布式锁服务。  
（1）zookeeper支持watcher机制，这样实现阻塞锁，可以watch锁数据，等到数据被删除， zookeeper会通知客户端去重新竞争锁。  
（2）zookeeper的数据可以支持临时节点的概念，即客户端写入的数据是临时数据，在客户端宕机后，临时数据会被删除，这样就实现了锁的异常释放。使用这样的方式，就不需要给锁增加超时自动释放的特性了。  
zookeeper实现锁的方式有两种：  
（1）客户端一起竞争写某条数据，比如/path/lock，只有第一个客户端能写入成功，其他的客户端都会写入失败。写入成功的客户端就获得了锁，写入失败的客户端，注册watch事件，等待锁的释放，从而继续竞争该锁。  
（2）客户端一起写入数据（创建自己的节点，临时顺序节点），所以每个节点的名字里都有序号，然后检查自己的序号是否是最小的，如果是成功获取锁，否则阻塞并添加一个watch监控自己前面的那个节点是否存在，当一个节点释放锁时会删除节点，这是会通知监控该节点状态的节点（下个节点），节点被激活之后去尝试获取锁（check自己的ID是不是最小的），如此类推。  
**优缺点：**  
\- 集群中的节点有一个leader，客户端只要从leader获取锁，其他节点能同步leader的数据，这样使用强一致性的分布式协调服务，分区、超时、冲突等问题都不会存在。所以为了保证分布。

下面是Zookeeper分布式锁的简单例子：

package com.zoo.example;
    
    import java.util.List;
    import java.util.TreeSet;
    import java.util.concurrent.Semaphore;
    
    import org.apache.log4j.Logger;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooDefs;
    import org.apache.zookeeper.ZooKeeper;
    
    import com.zoo.util.ConnectionUtil;
    
    /** * * @author Tim * */
    public class Locks { 
        public static final Logger logger = Logger.getLogger(Locks.class);
        /** * 这个信号量的作用是让获取锁的线程阻塞，这也是分布式锁设计的一个特性，否则线程获取不到锁立即失败返回 */
        private Semaphore semaphore = new Semaphore(1);
    
        private static ThreadLocal<String> myZnode = new ThreadLocal<String>();
        private ZooKeeper zk =  
        ConnectionUtil.connect("172.23.27.1:2181,172.23.27.2:2181,172.23.27.3:2181", 
                               20000,
                               semaphore);
        private static String root = "/locks";
        private static String separator = "/";
    
        /** * 尝试创建Zookeeper节点，从顺序值最小的节点开始依次获取锁，节点删除（释放锁），通过Zookeeper监听机制会唤醒 * 下一个节点，节点的顺序代表获取锁的顺序，节点的类型是 CreateMode.EPHEMERAL_SEQUENTIAL。 * @return * @throws InterruptedException * @throws KeeperException */
        boolean tryAquire() throws InterruptedException, KeeperException {
            semaphore.acquire();
    
            myZnode.set(zk.create(root + separator, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL));
    
            return acquireQueued();
        }
    
        /** * 获取失败，监听前继节点，阻塞等待唤醒 * @return * @throws KeeperException * @throws InterruptedException */
        boolean acquireQueued() throws KeeperException, InterruptedException {
    
            List<String> list = zk.getChildren(root, false);        
            TreeSet<String> allNodes = new TreeSet<String>(list);
    
            if (!myZnode.get().equals(root + "/" + allNodes.first())) {
    
                /** * 在当前节点的前继节点上添加监视器，节点删除时会往客户端发送一个监听事件， * 这个事件只会被监听这个节点的节点（后继节点）的监视器捕获，监视节点代表的线程获得信号量被唤醒 */
                if(null != zk.exists( root + "/" + String.format("%05d",                                                                                               
                Integer.parseInt(myZnode.get().substring(
                myZnode.get().lastIndexOf(separator)+1))-1), 
                new Watcher(){
                    @Override
                    public void process(WatchedEvent event){
                        if(event.getType() == Event.EventType.NodeDeleted){
                            System.out.println(Thread.currentThread().getName()+"节点被删除！");
                            semaphore.release();
                        }
                    }
                })){
                    semaphore.acquire();
                }
            }
            return true;
        }
    
        /** * 删除节点，释放锁 */
        void release() {
            try {
                zk.delete(myZnode.get(), -1);
    
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            }
        }
    
    }

**5. Curator**  
Curator提供了一套Java类库， 可以更容易的使用ZooKeeper。 ZooKeeper本身提供了JavaClient的访问类，但是API太底层，不宜使用， 容易出错。 Curator提供了三个组件。 Curatorclient用来替代ZOoKeeper提供的类， 它封装了底层的管理并提供了一些有用的工具。Curator framework提供了高级的API来简化ZooKeeper的使用。它增加了很多基于ZooKeeper的特性，帮助管理ZooKeeper的连接以及重试操作。Curator Recipes提供了使用ZooKeeper的一些通用的技巧（方法）。 除此之外， Curator Test提供了基于ZooKeeper的单元测试工具。所谓技巧(Recipes)，也可以称之为解决方案， 或者叫实现方案， 是指ZooKeeper的使用方法， 比如分布式的配置管理， Leader选举等。提供了fluent编程模型，提供了master选举，分布式锁，分布式基数，分布式barrier，可以很方便的为日常生产所使用。

public class CuratorLock {
                   public static CuratorFramework curator(){
                   String servers = "172.77.77.77:2181,172.77.77.78:2181,172.77.77.79:2181";
                   CuratorFramework curator = CuratorFrameworkFactory.builder().retryPolicy(new                   
                                          ExponentialBackoffRetry(10000,3)).connectString(servers).build();
                   curator.start();
                   return curator;
               }
               public static void main(String[] args) {
                   final InterProcessMutex lock = new InterProcessMutex(CuratorLock.curator(), "/dlock");
    
                   Executor pool = Executors.newFixedThreadPool(10);
                   for (int i = 0; i < 10; i ++) {
                       pool.execute(new Runnable() {
                       @Override
                       public void run() {
                            try {
                                 System.out.println("trying to acquire lock!");
                                 lock.acquire();
                                 System.out.println(Thread.currentThread().getName());
                                 TimeUnit.SECONDS.sleep(5);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }finally{
                                try {
                                    lock.release();
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            }
                        }
                     });
                 }
             }
           }