分布式式锁

背景

举例，比如当A,B,c三个人使用购买商品服务的时候，库存只有2。他们三个人同时发送购买请求，导致库存减3，库存变为-1。这种方式就是不加锁导致的。一般我们加锁的方式就是synchronize,lock等。但是这种只能锁当前进程内。对于现在分布式服务的环境下依旧存在问题。所以引入分布式锁。

分布式锁介绍

分布式锁最常用的有Redisson,zookeeper,Mysql等。单服务锁是单个进程使用锁标记。分布式锁是多个进程共用锁标记。

设计锁需要考虑的一些内容

• 排他性：在分布式服务集群中,同一个方法只能在一个服务一个机器上执行
• 容错性：分布式锁一定要等到释放,比如服务申请锁的过程中,服务器宕机/网络中断。这些都不能影响锁的释放
• 满足可重入、高性能、高应用
• 注意分布式锁的开销与颗粒度。

实现分布式锁的一些方式(了解就行，重点是Redisson)

基于数据库的实现方式

在数据库中创建一个表，表中包含方法名等字段，并在方法名字段上创建唯一索引，想要执行某个方法，就使用这个方法名向表中插入数据，成功插入则获取锁，执行完成后删除对应的行数据释放锁。

基于Redis的实现方式

获取锁的时候，使用setnx加锁，并使用expire命令为锁添加一个超时时间，超过该时间则自动释放锁，锁的value值为一个随机生成的UUID，通过此在释放锁的时候进行判断。

获取锁的时候还设置一个获取的超时时间，若超过这个时间则放弃获取锁。

释放锁的时候，通过UUID判断是不是该锁，若是该锁，则执行delete进行锁释放。

基于ZooKeeper的实现方式

（1）创建一个目录mylock；
（2）线程A想获取锁就在mylock目录下创建临时顺序节点；
（3）获取mylock目录下所有的子节点，然后获取比自己小的兄弟节点，如果不存在，则说明当前线程顺序号最小，获得锁；
（4）线程B获取所有节点，判断自己不是最小节点，设置监听比自己次小的节点；
（5）线程A处理完，删除自己的节点，线程B监听到变更事件，判断自己是不是最小的节点，如果是则获得锁。

这种方式很优雅，但是代码复杂，性能没有redis高

分布式锁最佳实践-Redisson

Redisson介绍

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。（官网抄的解释说明）。

官网文档地址https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95

Springboot使用redisson

创建一个redisson客户端

@Configuration
@Data
public class AppConfig {

    @Value("${spring.redis.host}")
    private String redisHost;

    @Value("${spring.redis.port}")
    private String redisPort;

    @Value("${spring.redis.password}")
    private String redisPwd;


    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        //单机
        config.useSingleServer().setPassword(redisPwd)
                .setTimeout(1000)
                .setRetryAttempts(3)
                .setRetryInterval(1000)
                .setPingConnectionInterval(1000) //**此项务必设置为redisson解决之前bug的timeout问题关键*****
                .setAddress("redis://"+redisHost+":"+redisPort);

        //集群
        //config.useClusterServers().setScanInterval(2000).addNodeAddress("redis://10.0.29.30:6379", "redis://10.0.29.95:6379");

        RedissonClient redissonClient = Redisson.create(config);
        return redissonClient;
    }
}

服务中使用

//引入客户端
@Autowired
private RedissonClient redissonClient;

public void serverSync(String couponId) {
  //建立锁标识
  String lockKey = "lock:coupon:"+couponId;
  
  //加锁
  RLock rlock = redissonClient.getLock(lockKey);
  rlock.lock();
  //
  try{
     //业务代码执行
  }finally{
     //解锁
     rlock.unlock();
  }
}

redisson解决分布式锁的坑

问题 : 锁的过期时间⼩于业务的执⾏时间该如何续期？

watch dog看⻔狗机制指定加锁时间

负责储存这个分布式锁的Redisson节点宕机以后，⽽且这个
锁正好处于锁住的状态时，这个锁会出现锁死的状态。或者业
务执⾏时间过⻓导致锁过期。

为了避免这种情况的发⽣， Redisson内部提供了⼀个监控锁
的看⻔狗，它的作⽤是在Redisson实例被关闭前，不断的延
⻓锁的有效期。

Redisson中客户端⼀旦加锁成功，就会启动⼀个watch
dog看⻔狗。 watch dog是⼀个后台线程，会每隔10秒检查
⼀下，如果客户端还持有锁key，那么就会不断的延⻓锁key
的⽣存时间默认情况下，看⻔狗的检查锁的超时时间是30秒钟，也可以
通过修改Config.lockWatchdogTimeout来另⾏指定

指定加锁时间

// 加锁以后10秒钟⾃动解锁
// ⽆需调⽤unlock⽅法⼿动解锁，没有watch dog
lock.lock(10, TimeUnit.SECONDS);

// 尝试加锁，最多等待100秒，上锁以后10秒⾃动解锁
boolean res = lock.tryLock(100, 10,
TimeUnit.SECONDS);
if (res) {
  try {
    ...
  } finally {
    lock.unlock();
  }
}

更多的redisson可以在官网上找寻使用方式,例如redisson公平锁等。

欢迎关注「后端技术小宅圈」一起交流，一起成为大牛