微服务分布式锁解决方案

旧城等待， 2022-04-06 11:12 279阅读 0赞

下面说一下分布式实现的几种方式：

一、数据库悲观锁

所谓的悲观锁：顾名思义，就是很悲观，每次去拿数据的时候都认为别人会修改，所以每次拿数据的时候都会上锁。这样别人拿数据的时候就要等待直到锁的释放。

这里是采用oracle的 select  ......  where id=1 for update 来实现分布式锁，建议加上nowait，或者wait 以及 of

下面是demo：

select * from table where id=1 for update nowait;//当锁被占用，不等待直接报错
    select * from table where id=1 for update wait 6;//当锁被占用，等待6s
    select * from table where id=1 for update of columns nowait;//锁定执行的列，反之则锁所有列。

该方案，在高并发时显然不适用，依赖于数据库的性能以及锁机制，会造成锁无法释放。

二、数据库乐观锁

所谓的乐观锁：就是很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据。一般的方案都是加一个版本号字段（version），在查询数据时将版本号带出来，更新后将版本号+1，如果版本号一致才更新，并获取影响行数，如果没更新则报错。

三、redis的setnx

由于redis是单线程工作的，所以它存取key-value 的时候是单线程工作的，并且多个线程请求redis并不存在竞争问题。所以可以设置一个标识来作为一把锁，只有获取了该锁之后，才能对共享的资源进行操作，没有拿到锁的线程处于不断去取锁的状态，直到等到上一个线程释放锁（即后一个线程可以取到锁）或者超过规定超时时间不再取锁。

import com.test.core.base.utils.ApplicationUtil;
    import com.test.redis.api.RedisStringOperationService;
    
    /**
     * redis分布式锁
     * @author LIPENG
     * @date 2017年9月22日 下午4:25:56
     * @version V1.0
     */
    public class RedisDistributionLock {
        private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;
        /**
         * Lock key path.
         */
        private String lockKey;
    
        /**
         * 锁超时时间，防止线程在入锁以后，无限的执行等待
         */
        private int expireMsecs = 60 * 1000;
    
        /**
         * 锁等待时间，防止线程饥饿
         */
        private int timeoutMsecs = 10 * 1000;
    
        private volatile boolean locked = false;
    
        /**
         * Detailed constructor with default acquire timeout 10000 msecs and lock expiration of 60000 msecs.
         *
         * @param lockKey lock key (ex. account:1, ...)
         */
        public RedisDistributionLock(String lockKey) {
            this.lockKey = lockKey + "_lock";
        }
    
        /**
         * Detailed constructor with default lock expiration of 60000 msecs.
         *
         */
        public RedisDistributionLock(String lockKey, int timeoutMsecs) {
            this(lockKey);
            this.timeoutMsecs = timeoutMsecs;
        }
    
        /**
         * Detailed constructor.
         *
         */
        public RedisDistributionLock(String lockKey, int timeoutMsecs, int expireMsecs) {
            this(lockKey, timeoutMsecs);
            this.expireMsecs = expireMsecs;
        }
    
        /**
         * @return lock key
         */
        public String getLockKey() {
            return lockKey;
        }
    
    
    
        /**
         * 获得 lock.
         * 实现思路: 主要是使用了redis 的setnx命令,缓存了锁.
         * reids缓存的key是锁的key,所有的共享, value是锁的到期时间(注意:这里把过期时间放在value了,没有时间上设置其超时时间)
         * 执行过程:
         * 1.通过setnx尝试设置某个key的值,成功(当前没有这个锁)则返回,成功获得锁
         * 2.锁已经存在则获取锁的到期时间,和当前时间比较,超时的话,则设置新的值
         *
         * @return true if lock is acquired, false acquire timeouted
         * @throws InterruptedException in case of thread interruption
         */
        public synchronized boolean lock() throws InterruptedException {
            int timeout = timeoutMsecs;
            while (timeout >= 0) {
                long expires = System.currentTimeMillis() + expireMsecs + 1;
                String expiresStr = String.valueOf(expires); //锁到期时间
                if (getRedisStringOperationService().setNX(lockKey, expiresStr)) {
                    // lock acquired
                    locked = true;
                    return true;
                }
    
                String currentValueStr = getRedisStringOperationService().get(lockKey); //redis里的时间
                if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
                    //判断是否为空，不为空的情况下，如果被其他线程设置了值，则第二个条件判断是过不去的
                    // lock is expired
    
                    String oldValueStr = getRedisStringOperationService().getAndSet(lockKey, expiresStr);
                    //获取上一个锁到期时间，并设置现在的锁到期时间，
                    //只有一个线程才能获取上一个线上的设置时间，因为jedis.getSet是同步的
                    if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
                        //防止误删（覆盖，因为key是相同的）了他人的锁——这里达不到效果，这里值会被覆盖，但是因为什么相差了很少的时间，所以可以接受
    
                        //[分布式的情况下]:如过这个时候，多个线程恰好都到了这里，但是只有一个线程的设置值和当前值相同，他才有权利获取锁
                        // lock acquired
                        locked = true;
                        return true;
                    }
                }
                timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;
    
                /*
                    延迟100 毫秒,  这里使用随机时间可能会好一点,可以防止饥饿进程的出现,即,当同时到达多个进程,
                    只会有一个进程获得锁,其他的都用同样的频率进行尝试,后面有来了一些进行,也以同样的频率申请锁,这将可能导致前面来的锁得不到满足.
                    使用随机的等待时间可以一定程度上保证公平性
                 */
                Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);
    
            }
            return false;
        }
    
    
        /**
         * 释放锁
         */
        public synchronized void unlock() {
            if (locked) {
            	getRedisStringOperationService().delete(lockKey);
                locked = false;
            }
        }
        
        private RedisStringOperationService getRedisStringOperationService(){
        	return	ApplicationUtil.getBean(RedisStringOperationService.class);
        }
    
    }

调用方式：

RedisDistributionLock lock = new RedisDistributionLock("key", 10000, 20000);
    		try {
    			if (lock.lock()) {
    				// 需要加锁的代码
    			}
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} finally {
    			lock.unlock();
    		}

四、使用zookeeper

当很多进程需要访问共享资源时，我们可以通过zk来实现分布式锁。主要步骤是：   
1.建立一个节点，假如名为：lock 。节点类型为持久节点（PERSISTENT）   
2.每当进程需要访问共享资源时，会调用分布式锁的lock()或tryLock()方法获得锁，这个时候会在第一步创建的lock节点下建立相应的顺序子节点，节点类型为临时顺序节点（EPHEMERAL\_SEQUENTIAL），通过组成特定的名字name+lock+顺序号。   
3.在建立子节点后，对lock下面的所有以name开头的子节点进行排序，判断刚刚建立的子节点顺序号是否是最小的节点，假如是最小节点，则获得该锁对资源进行访问。   
4.假如不是该节点，就获得该节点的上一顺序节点，并给该节点是否存在注册监听事件。同时在这里阻塞。等待监听事件的发生，获得锁控制权。   
5.当调用完共享资源后，调用unlock（）方法，关闭zk，进而可以引发监听事件，释放该锁。   
实现的分布式锁是严格的按照顺序访问的并发锁。

package cn.wpeace.zktest;
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.Collections;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    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 org.apache.zookeeper.Watcher.Event.KeeperState;
    import org.apache.zookeeper.data.Stat;
    /**
     * @author peace
     *
     */
    public class DistributedLock implements Lock, Watcher{
        private ZooKeeper zk;
        private String root = "/locks";//根
        private String lockName;//竞争资源的标志
        private String waitNode;//等待前一个锁
        private String myZnode;//当前锁
        private CountDownLatch latch;//计数器
        private CountDownLatch connectedSignal=new CountDownLatch(1);
        private int sessionTimeout = 30000; 
        /**
         * 创建分布式锁,使用前请确认config配置的zookeeper服务可用
         * @param config 192.168.1.127:2181
         * @param lockName 竞争资源标志,lockName中不能包含单词_lock_
         */
        public DistributedLock(String config, String lockName){
            this.lockName = lockName;
            // 创建一个与服务器的连接
             try {
                zk = new ZooKeeper(config, sessionTimeout, this);
                connectedSignal.await();
                Stat stat = zk.exists(root, false);//此去不执行 Watcher
                if(stat == null){
                    // 创建根节点
                    zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); 
                }
            } catch (IOException e) {
                throw new LockException(e);
            } catch (KeeperException e) {
                throw new LockException(e);
            } catch (InterruptedException e) {
                throw new LockException(e);
            }
        }
        /**
         * zookeeper节点的监视器
         */
        public void process(WatchedEvent event) {
            //建立连接用
            if(event.getState()==KeeperState.SyncConnected){
                connectedSignal.countDown();
                return;
            }
            //其他线程放弃锁的标志
            if(this.latch != null) {  
                this.latch.countDown();  
            }
        }
    
        public void lock() {   
            try {
                if(this.tryLock()){
                    System.out.println("Thread " + Thread.currentThread().getId() + " " +myZnode + " get lock true");
                    return;
                }
                else{
                    waitForLock(waitNode, sessionTimeout);//等待锁
                }
            } catch (KeeperException e) {
                throw new LockException(e);
            } catch (InterruptedException e) {
                throw new LockException(e);
            } 
        }
        public boolean tryLock() {
            try {
                String splitStr = "_lock_";
                if(lockName.contains(splitStr))
                    throw new LockException("lockName can not contains \\u000B");
                //创建临时子节点
                myZnode = zk.create(root + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
                System.out.println(myZnode + " is created ");
                //取出所有子节点
                List<String> subNodes = zk.getChildren(root, false);
                //取出所有lockName的锁
                List<String> lockObjNodes = new ArrayList<String>();
                for (String node : subNodes) {
                    String _node = node.split(splitStr)[0];
                    if(_node.equals(lockName)){
                        lockObjNodes.add(node);
                    }
                }
                Collections.sort(lockObjNodes);
    
                if(myZnode.equals(root+"/"+lockObjNodes.get(0))){
                    //如果是最小的节点,则表示取得锁
                    System.out.println(myZnode + "==" + lockObjNodes.get(0));
                    return true;
                }
                //如果不是最小的节点，找到比自己小1的节点
                String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1);
                waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, subMyZnode) - 1);//找到前一个子节点
            } catch (KeeperException e) {
                throw new LockException(e);
            } catch (InterruptedException e) {
                throw new LockException(e);
            }
            return false;
        }
        public boolean tryLock(long time, TimeUnit unit) {
            try {
                if(this.tryLock()){
                    return true;
                }
                return waitForLock(waitNode,time);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return false;
        }
        private boolean waitForLock(String lower, long waitTime) throws InterruptedException, KeeperException {
            Stat stat = zk.exists(root + "/" + lower,true);//同时注册监听。
            //判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听
            if(stat != null){
                System.out.println("Thread " + Thread.currentThread().getId() + " waiting for " + root + "/" + lower);
                this.latch = new CountDownLatch(1);
                this.latch.await(waitTime, TimeUnit.MILLISECONDS);//等待，这里应该一直等待其他线程释放锁
                this.latch = null;
            }
            return true;
        }
        public void unlock() {
            try {
                System.out.println("unlock " + myZnode);
                zk.delete(myZnode,-1);
                myZnode = null;
                zk.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            }
        }
        public void lockInterruptibly() throws InterruptedException {
            this.lock();
        }
        public Condition newCondition() {
            return null;
        }
    
        public class LockException extends RuntimeException {
            private static final long serialVersionUID = 1L;
            public LockException(String e){
                super(e);
            }
            public LockException(Exception e){
                super(e);
            }
        }
    }

调用方法：

DistributedLock lock   = new DistributedLock("192.168.1.127:2181","lock");
     lock.lock();
     //共享资源
     if(lock != null)
      lock.unlock();