Java分布式锁-阿里云开发者社区

Java分布式锁

2024-10-14 7

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： Java分布式锁

分布式锁是一种在分布系统环境下，通过多个节点对共享资源进行访问控制的一种同步机制。

主要的目的是防止多个节点同时操作同一份数据，从而避免数据的不一致性。分布式锁的实现比线程锁和进程锁要复杂得多，因为它需要在网诺中的多个节点之间进行协调，以保证锁的唯一性和一致性。

分布式锁的基本原理可以分为以下几个步骤：

请求锁：当一个实例需要访问共享资源时，它会向分布式系统发送一个请求，试图获取一个锁。
锁定资源：分布式锁系统会检查是否有其他实例已经有这个锁，如果没有，那么这个锁实例就会获得锁，并且有权访问共享资源，如果有那么这个实例就必须等待，直到被释放。
访问资源：一旦实例获取锁，它就会安全访问共享资源，而不用担心其他实例会同时访问这个资源。
释放资源：当实例完成对共享资源的访问后，它需要通知分布式锁释放锁，这样其他正在等待的实例就可以获取锁，访问共享资源。

分布式锁的实现方式

在实现分布式锁时，通常会有一个中心节点（或者称为锁服务），所有需要获取锁的节点都需要向这个中心节点申请。这个中心节点负责协调和管理所有节点的锁请求，确保锁的唯一性和一致性。

分布式锁的特性

分布式锁主要有以下几个特性：

互斥性：在任何时刻，只有一个节点可以持有锁。
不会发生死锁：如果一个节点崩溃，锁可以被其他节点获取。
公平性：如果多个节点同时申请锁，系统应该保证每个节点都有获取锁的机会。
可重入性：同一个节点可以多次获取同一个锁，而不会被阻塞。
高可用：锁服务应该是高可用的，不能因为锁服务的故障而影响整个系统的运行

java中实现分布式锁的常见方式有以下三种

1.使用数据库实现分布锁

悲观锁

利用select … where … for update 排他锁

注意: 其他附加功能与实现一基本一致，这里需要注意的是“where name=lock ”，name字段必须要走索引，否则会锁表。有些情况下，比如表不大，mysql优化器会不走这个索引，导致锁表问题。

乐观锁

所谓乐观锁与前边最大区别在于基于CAS思想，是不具有互斥性，不会产生锁等待而消耗资源，操作过程中认为不存在并发冲突，只有update version失败后才能觉察到。我们的抢购、秒杀就是用了这种实现以防止超卖。

通过增加递增的版本号字段实现乐观锁

2.使用Redis实现分布式锁

根据系统的业务设置唯一值，用于解锁验证

1 /**
  2  * 分布式锁的简单实现代码  4  */
  5 public class DistributedLock {
  6 
  7     private final JedisPool jedisPool;
  8 
  9     public DistributedLock(JedisPool jedisPool) {
 10         this.jedisPool = jedisPool;
 11     }
 12 
 13     /**
 14      * 加锁
 15      * @param lockName       锁的key
 16      * @param acquireTimeout 获取超时时间
 17      * @param timeout        锁的超时时间
 18      * @return 锁标识
 19      */
 20     public String lockWithTimeout(String lockName, long acquireTimeout, long timeout) {
 21         Jedis conn = null;
 22         String retIdentifier = null;
 23         try {
 24             // 获取连接
 25             conn = jedisPool.getResource();
 26             // 随机生成一个value
 27             String identifier = UUID.randomUUID().toString();
 28             // 锁名，即key值
 29             String lockKey = "lock:" + lockName;
 30             // 超时时间，上锁后超过此时间则自动释放锁
 31             int lockExpire = (int) (timeout / 1000);
 32 
 33             // 获取锁的超时时间，超过这个时间则放弃获取锁
 34             long end = System.currentTimeMillis() + acquireTimeout;
 35             while (System.currentTimeMillis() < end) {
 36                 if (conn.setnx(lockKey, identifier) == 1) {
 37                     conn.expire(lockKey, lockExpire);
 38                     // 返回value值，用于释放锁时间确认
 39                     retIdentifier = identifier;
 40                     return retIdentifier;
 41                 }
 42                 // 返回-1代表key没有设置超时时间，为key设置一个超时时间
 43                 if (conn.ttl(lockKey) == -1) {
 44                     conn.expire(lockKey, lockExpire);
 45                 }
 46 
 47                 try {
 48                     Thread.sleep(10);
 49                 } catch (InterruptedException e) {
 50                     Thread.currentThread().interrupt();
 51                 }
 52             }
 53         } catch (JedisException e) {
 54             e.printStackTrace();
 55         } finally {
 56             if (conn != null) {
 57                 conn.close();
 58             }
 59         }
 60         return retIdentifier;
 61     }
 62 
 63     /**
 64      * 释放锁
 65      * @param lockName   锁的key
 66      * @param identifier 释放锁的标识
 67      * @return
 68      */
 69     public boolean releaseLock(String lockName, String identifier) {
 70         Jedis conn = null;
 71         String lockKey = "lock:" + lockName;
 72         boolean retFlag = false;
 73         try {
 74             conn = jedisPool.getResource();
 75             while (true) {
 76                 // 监视lock，准备开始事务
 77                 conn.watch(lockKey);
 78                 // 通过前面返回的value值判断是不是该锁，若是该锁，则删除，释放锁
 79                 if (identifier.equals(conn.get(lockKey))) {
 80                     Transaction transaction = conn.multi();
 81                     transaction.del(lockKey);
 82                     List<Object> results = transaction.exec();
 83                     if (results == null) {
 84                         continue;
 85                     }
 86                     retFlag = true;
 87                 }
 88                 conn.unwatch();
 89                 break;
 90             }
 91         } catch (JedisException e) {
 92             e.printStackTrace();
 93         } finally {
 94             if (conn != null) {
 95                 conn.close();
 96             }
 97         }
 98         return retFlag;
 99     }
100 }

3.使用Zookeeper实现分布式锁

创建临时节点，执行业务逻辑，释放锁。

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
 
/**
 * 分布式锁Zookeeper实现
 *
 */
@Slf4j
@Component
public class ZkLock implements DistributionLock {
private String zkAddress = "zk_adress";
    private static final String root = "package root";
    private CuratorFramework zkClient;
 
    private final String LOCK_PREFIX = "/lock_";
 
    @Bean
    public DistributionLock initZkLock() {
        if (StringUtils.isBlank(root)) {
            throw new RuntimeException("zookeeper 'root' can't be null");
        }
        zkClient = CuratorFrameworkFactory
                .builder()
                .connectString(zkAddress)
                .retryPolicy(new RetryNTimes(2000, 20000))
                .namespace(root)
                .build();
        zkClient.start();
        return this;
    }
 
    public boolean tryLock(String lockName) {
        lockName = LOCK_PREFIX+lockName;
        boolean locked = true;
        try {
            Stat stat = zkClient.checkExists().forPath(lockName);
            if (stat == null) {
                log.info("tryLock:{}", lockName);
                stat = zkClient.checkExists().forPath(lockName);
                if (stat == null) {
                    zkClient
                            .create()
                            .creatingParentsIfNeeded()
                            .withMode(CreateMode.EPHEMERAL)
                            .forPath(lockName, "1".getBytes());
                } else {
                    log.warn("double-check stat.version:{}", stat.getAversion());
                    locked = false;
                }
            } else {
                log.warn("check stat.version:{}", stat.getAversion());
                locked = false;
            }
        } catch (Exception e) {
            locked = false;
        }
        return locked;
    }
 
    public boolean tryLock(String key, long timeout) {
        return false;
    }
 
    public void release(String lockName) {
        lockName = LOCK_PREFIX+lockName;
        try {
            zkClient
                    .delete()
                    .guaranteed()
                    .deletingChildrenIfNeeded()
                    .forPath(lockName);
            log.info("release:{}", lockName);
        } catch (Exception e) {
            log.error("删除", e);
        }
    }
 
    public void setZkAddress(String zkAddress) {
        this.zkAddress = zkAddress;
    }
}