Redisson入门 - 《Java？Java！》

1.什么是Redisson?
2.Redisson用于干什么？
3.分布式锁
4.Redisson入门
5.Redis主从架构锁失效
6.RedLock实现
7.用注解实现Redis分布式锁
8.后续遗留问题

1.什么是Redisson?

Redisson是架设在Redis基础上的一个Java驻内存数据网格（In-Memory Data Grid），也是官方推荐的分布式锁解决方案。

Redisson在基于NIO的Netty框架上，充分的利用了Redis键值数据库提供的一系列优势，在Java实用工具包中常用接口的基础上，为使用者提供了一系列具有分布式特性的常用工具类。使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度。同时结合各富特色的分布式服务，更进一步简化了分布式环境中程序相互之间的协作。

兼容 Redis 2.6+ and JDK 1.6+，使用Apache License 2.0授权协议。

2.Redisson用于干什么？

场景：秒杀、抢优惠卷、接口的幂等校验等……

简单模拟一个超卖的情况

package com.zym.redisson.test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class IndexController {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @RequestMapping("/stock")
    public String stock(){
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if (stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",String.valueOf(realStock));
            System.out.println("购买成功，剩余库存：" + realStock);
        }else {
            System.out.println("购买失败，库存不足");
        }
        return "end";
    }
    @RequestMapping("/add")
    public void add(){
        stringRedisTemplate.opsForValue().set("stock","100");
        System.out.println("添加成功！");
    }
}

我们试想一下，如果在高并发的情况下会出现什么问题？假设同时有三个人买东西，100-3=97。但实际上，我们会看到剩余库存为99，这就是超卖了。

老板：不是定好了卖100件吗，怎么一夜之间多了300多个订单？

程序员：……

如何优化呢？简单，我们可以用synchronized把代码包裹起来。是的，这样是完全可行的，但我们也要注意，synchronized解决的并发问题，仅限于解决单机部署系统的并发问题，如果你的项目是多节点部署，并发问题依然存在（使用jmeter工具很容易就可以测试出来），其次，synchronized作为一个重量级锁，性能也会出现很大的问题，综上所诉，synchronized在秒杀环境下并不实用。

3.分布式锁

那么我们用什么可以避免秒杀场景下的超卖问题呢？其实在redis内部的一条命令，就实现了分布式锁：SETNX（set if not null）

该命令为，当且仅当key不存在时，才会写入redis。

我们对上面的例子进行改造：

package com.zym.redisson.test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class IndexController {
   @Autowired
   private StringRedisTemplate stringRedisTemplate;
   @RequestMapping("/stock")
   public String stock() {
       Boolean result = stringRedisTemplate.opsForValue().setIfAbsent("lockKey", "true");
       if (!result) {
           return "error";
       }
       int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
       if (stock > 0) {
           int realStock = stock - 1;
           stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
           System.out.println("购买成功，剩余库存：" + realStock);
       } else {
           System.out.println("购买失败，库存不足");
       }
       stringRedisTemplate.delete("lockKey");
       return "end";
   }
   @RequestMapping("/add")
   public void add() {
       stringRedisTemplate.opsForValue().set("stock", "100");
       System.out.println("添加成功！");
   }
}

这样我们就简单的实现了一个入门级的分布式锁。但是这样就没有问题了吗？

假设我加锁成功后出现了异常（抛异常、服务挂掉、redis挂掉），执行完业务后并未释放锁，那么这样的话，后续用户都无法正常去买东西了

月黑风高的晚上，用户们静静等待12点的到来，以便于去抢购商品，抢着抢着发现抢不到，在看库存还有80多个，气的用户大骂：什么垃圾系统！

该场景模拟抛出异常导致未释放锁造成了死锁，那么这样的话，我们就需要把我们必须执行的代码放到finally中来避免服务挂掉的死锁。

运维：诶，有个紧急bug，我要赶紧发布个版本（正好是秒杀时间）

运维：诶，怎么后台没有订单增加了，开发，开发呢！

开发：……

在服务挂掉的情况下，我们就需要在加锁的时候设置超时时间，防止服务挂掉导致的死锁。

经过我们一番优化后，看起来超卖的问题已经被我们解决了。但是，我们这里还会有一点问题，当我们加锁成功后，因为网络问题，导致在请求数据库时耗时很长，以至于redis释放了锁，又因为是高并发，导致第二个请求加上了锁去执行，然而请求一也还在执行，请求一执行完成后告诉redis，可以释放锁了。但是现在这把锁是请求二锁着的，然后redis释放了锁，请求三上锁……甚至导致锁永久失效（刚加锁就释放）

开发：懂了，我马上收拾铺盖走人。

4.Redisson入门

那么对于上面这种情况，我们可以给每一个线程设置一个id，在释放锁的时候，判断一下是不是当前线程的id，如果是，则释放，若不是，则不释放。

但是我们也要注意，这样解决的问题是锁永久失效的问题，若请求一因为网络慢的问题导致redis锁释放，那么还是会有超卖问题。

其实市面上已经有了一套用来解决分布式锁的方案，那就是Redisson，简单改造一下代码

public String stock() {
        RLock redissonLock = redisson.getLock("lockKey");
        redissonLock.lock(5,TimeUnit.SECONDS);
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if (stock > 0) {
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock", String.valueOf(realStock));
            System.out.println("购买成功，剩余库存：" + realStock);
        } else {
            System.out.println("购买失败，库存不足");
        }
        redissonLock.unlock();
        return "end";
    }

我们简单分析一下源码：

我们可以看到一段lua脚本

然后我们在看一下那个定时任务，也叫看门狗：

我们追踪这段代码：

这个脚本也好理解，延长了key的时间

那么每隔10秒执行一次的时间是在哪里设置的呢？

5.Redis主从架构锁失效

当我们的redis用的是主从架构的话，在极端情况下还是会有问题。

我们知道，主节点同步从节点是异步的，那么如果主节点在同步的时候挂掉，从节点就无法同步这个key，然后从节点被推举为主节点，第二个请求进来发现没有key，加锁进行业务逻辑，而请求一时间较长，那么这种时候，就又会出现问题了。

这种情况，我们可能就要考虑用zookeeper来解决分布式锁的问题了。

cap原则：c（Consistency）：一致性。a（Availability）：可用性。p（Partition tolerance）：分区容错性

cap只能满足其中两项，无法满足三项

redis集群：ap

zookeeper集群：cp

当redis写入了新数据，程序立马继续执行，然后异步同步key到从机

而zookeeper必须要等到把所有的节点同步完成后，才继续执行程序

当主节点挂掉后，redis集群会进行选举，不知道哪个从机会变为主机，而zookeeper内部的ZAB协议会保证某个跟随者（follower）会变成领导者（leader）

那么如何选择呢？对并发要求高的话，就用redis。对一致性要求高的话，就用zookeeper。

那么我们不能使用redis解决主从架构锁失效的问题吗？

可以，这时我们要用到redlock。

6.RedLock实现

RLock lock1 = redisson1.getLock("lock1");
RLock lock2 = redisson2.getLock("lock2");
RLock lock3 = redisson3.getLock("lock3");
RLock redLock = anyRedisson.getRedLock(lock1, lock2, lock3);
// traditional lock method
redLock.lock();
// or acquire lock and automatically unlock it after 10 seconds
redLock.lock(10, TimeUnit.SECONDS);
// or wait for lock aquisition up to 100 seconds 
// and automatically unlock it after 10 seconds
boolean res = redLock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
   try {
     ...
   } finally {
       redLock.unlock();
   }
}

原文
Martin Kleppmann（分布式专家）的质疑贴
Antirez（Redis之父）的反击贴

7.用注解实现Redis分布式锁

AOP是一大利器，让我们可以在代码中专精业务，而不需要管额外的事情

DistributedLock:

import java.lang.annotation.*;
@Inherited //该注解的意思为：当@DistributedLock注解加在了类A中，假如类B继承了A，则B也会带上该注解。
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface DistributedLock {
    String key() default "redisLockKey";
    int waitTime() default 5;
    int expireTime() default 5;
}

DistributedLockAspect:

import com.zym.redisson.annotation.DistributedLock;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;
@Component
@Aspect
@Slf4j
public class DistributedLockAspect {
    @Autowired
    private RedissonClient redissonClient;
    @Around("@annotation(com.zym.redisson.annotation.DistributedLock)")
    public Object around(ProceedingJoinPoint joinPoint){
        Signature signature = joinPoint.getSignature();
        MethodSignature methodSignature = (MethodSignature) signature;
        Method targetMethod = methodSignature.getMethod();
        DistributedLock distributedLock = targetMethod.getDeclaredAnnotation(DistributedLock.class);
        String lockKey = distributedLock.key();
        RLock lock = redissonClient.getLock(lockKey);
        boolean locked = false;
        Object result = null;
        try{
            locked = lock.tryLock(distributedLock.waitTime(),distributedLock.expireTime(), TimeUnit.SECONDS);
            if (locked){
                //处理业务
                result = joinPoint.proceed();
            }else {
                log.info("请稍后重试");
                throw new Exception();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        }finally {
            if (locked){
                lock.unlock();
            }
        }
        return result;
    }
}

8.后续遗留问题

如何优化分布式锁性能？

源码地址：https://gitee.com/zym213/introduction-to-redisson.git