在什么网站可以做外贸出口劳保鞋/如何做线上推广

👨‍🎓作者简介：一位大四、研0学生，正在努力准备大四暑假的实习
🌌上期文章：Redis：原理速成+项目实战——Redis实战7（优惠券秒杀+细节解决超卖、一人一单问题）
📚订阅专栏：Redis：原理速成+项目实战
希望文章对你们有所帮助

上一篇文章已经通过代码的调优，用Redis实现了单个JVM下的秒杀并保证了线程安全问题，但是通过测试发现，在集群分布下，JVM之间依旧会存在线程安全问题，解决这个问题的方法就是分布式锁。

因为是速成，所以这一篇涉及到的底层的原理（Redisson的锁重试和WatchDog机制、Redisson的multiLock原理）只能讲个大概，但是他们的源码真的得太久了。。。把源码的实现做个总结也不太现实，还是需要大家自己去啃。（我从晚上11点啃到凌晨3点。。。）
另外这篇文章的最后一部分测试，我配置了多个Redis结点，自己去配置是很繁琐的，所以我会用Docker来进行配置，有关于Docker的文章可以看这：
一文快速学会Docker软件部署

分布式锁

基本原理

JVM内的线程之间可以用锁实现互斥，是因为一个他们的锁只有一个锁监视器，每个JVM都有一个锁监视器，但是多个JVM就会有多个锁监视器，导致发生线程安全问题。
因此，要实现互斥，可以让多个JVM都共用一个锁监视器，这样让JVM与JVM之间、每个JVM的线程之间都共用这个锁，就不会发生线程安全问题了。
由此引出分布式锁的定义：满足分布式系统或集群模式下多进程可见并且互斥的锁。
需要满足的特点：多进程可见、互斥、高可用、高性能、安全性

不同实现方式对比

	MySQL	Redis	Zookeeper
互斥	本身的互斥锁机制	利用互斥命令setnx	利用节点的唯一性和有序性实现互斥
高可用	好	好	好
高性能	一般	好	一般
安全性	断开连接，自动释放锁	利用锁超时时间，到时释放	临时节点，断开连接自动释放

基于Redis的分布式锁

之前讨论过，我们的方式就是用Redis中的setnx去设置一个锁，而为了解决锁释放前出现以外，我们会给锁增加一个超时释放expire，这样即便出现异常，也不会一直不释放，其他线程也能正常获得锁并执行操作。
获取锁：set lock thread1 NX EX 10（这里的expire就不要单独写一行了，要保持原子性，不然有可能expire还没执行Redis就宕机，照样会造成锁无法释放的情况）
释放锁：del key

需要讨论一下，其他线程获取锁失败以后该怎么办，我们选用非阻塞式的方式，当获取锁失败了以后，不再等待（成功返回true，否则返回false）

容易总结出流程：

实现Redis分布式锁初级

直接在utils包下创建ILock接口与SimpleRedisLock 类，这个内容和之前的差不多，用stringRedisTemplate完成的流程就那一套：

public class SimpleRedisLock implements ILock{public static final String KEY_PREFIX = "lock:";private String name;//不同业务有不同的锁，业务name即为锁的nameprivate StringRedisTemplate stringRedisTemplate;public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}@Overridepublic boolean tryLock(long timeoutSec) {//获取线程表示long threadId = Thread.currentThread().getId();//获取锁Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);//防止拆箱操作，不能直接返回successreturn Boolean.TRUE.equals(success);}@Overridepublic void unLock() {stringRedisTemplate.delete(KEY_PREFIX + name);}
}

接着修改我们的下单业务的impl，改变之前的加锁逻辑：

        //创建锁对象，key需要加上用户id，因为不同的用户无所谓，只有同一个用户才要锁起来，因此要指定好用户idSimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);//获取锁boolean isLock = lock.tryLock(1200);//判断是否获取锁成功if(!isLock){//获取锁失败return Result.fail("不允许重复下单");}//获取代理对象try {IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);} finally {//手动释放锁lock.unLock();}

在锁那打断点，并利用postman发请求就可以看到锁起到作用了，这都是基本功了。

Redis分布式锁误删问题

上面的锁已经可以解决大多数的情况了，但是遇到一些极端情况还是会出问题：
当一个线程的业务阻塞了，甚至到达了key的TTL，这时候就会被强制释放锁，因此其他的线程就可以成功获取锁并执行自己的业务，而一旦之前被阻塞的业务完成了自己的业务，并且去unLock，这时候就会释放了其它业务的锁，这时候就会导致本来在执行的业务没有了锁，再次引发安全问题。

这个情况出现的情况相对没有那么大，但是一旦出现就可能会大量出现并发安全问题，因此需要解决问题。
如上图，归根结底，发生大量线程并发问题的原因是线程1误删了线程2的锁，因此我们可以尝试进行一个资格判断，判断线程1此时有没有资格释放锁，这是解决误删问题的一个思路：

我们需要修改一下业务流程：

解决Redis分布式锁误删问题

根据上述的分析，我们需要修改一下分布式锁，使得满足：
1、在获取锁时存入线程标识

在这里增加了UUID来作为线程的标识，不再使用线程自己的ID了，这是因为虽然每个JVM的线程都是递增的，每个JVM内部之间的都会维护线程的唯一ID，但是不同的JVM之间还是会产生冲突，因此让JVM自己去维护线程的ID，会导致不同JVM之间的ID冲突。
事实上，也可以用UUID来表示不同的JVM，用线程ID来区分JVM内部的线程，两者拼接在一块。

2、在释放锁时限获取锁中的线程标识，判断是否与当前线程标识一致（一致才可释放）

业务内部，需要增加线程标识的prefix：

接着修改tryLock与unLock的逻辑，线程的标识变成UUID+线程ID

这样就可以解决不同JVM之间锁的误删问题，可自行DEBUG。
但这样做依旧不是完美方案。

分布式锁的原子性问题

上述的方式已经可以解决业务阻塞导致的误删操作，但是还会有一些问题：

如果我们阻塞的不是业务，而是业务执行完了，并且判断锁标识成功，即将释放锁的时候发生的阻塞（这种阻塞不是业务阻塞，而可能是JVM内部的垃圾回收机制异常导致阻塞），这时候还会发生新的问题。
如果被阻塞的时间足够长，导致锁的TTL到期了，一旦释放，其他线程又开始乘虚而入，成功获取锁，执行业务。
这时候，被阻塞的线程恢复正常了，但是因为已经进行锁标识的逻辑判断了，这时候被阻塞的线程就可以完成这个释放锁的操作，再次造成误删问题。

可以看下图：

分析一下问题发生的原因，之所以会出现这种情况，主要原因是锁标识的逻辑判断与锁的释放操作，是两个不同的操作，不满足原子性，所以当在两个操作之间发生了阻塞，那么线程并发问题依旧会出现。
所以，我们必须要保证判断锁标识的动作与释放锁的动作必须得保证原子性。

Lua脚本

想到原子性，我们很容易就想到MySQL中的事务，但是Redis中的事务却不太一样，Redis事务虽然能保障原子性，但是无法保证事务的一致性。Redis事务的操作是一系列的批处理，是在最终的一致性执行的，必须要有乐观锁来做判断，会麻烦很多。
Lua语言能够保证原子性，是因为它在执行原子操作时会将其他线程或进程阻塞，直到该操作完成。
而Redis提供了Lua脚本，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Lua是一种变成语言，基本语法可以参考：
Lua语法教程
重点介绍Lua中Redis提供的调用函数：

redis.call(‘命令名称’, ‘key’, ‘其它参数’, …)

例如，执行set name jack，脚本写法如下：

redis.call(‘set’, ‘name’, ‘jack’)

在我们编写完脚本，使得多条命令的操作满足了原子性，我们还需要用Redis命令来调用脚本：

EVAL script numkeys key… arg…

例如，要执行redis.call(‘set’, ‘name’, ‘jack’)这个脚本：

EVAL “return redis.call(‘set’, ‘name’, ‘jack’)” 0

0表示key类型的参数的个数

脚本中的key、value不要写死，那可以作为参数传递。key类型参数会放入KEYS数组，其它参数会放入ARGV十足，在脚本中可以从KEYS和ARGV数组获取这些参数：

EVAL “return redis.call(‘set’, KEYS[1], ARGV[1])” 1 name Rose
1代表key类型的参数有一个，也就是紧接着的name，会放入KEYS[1]
而Rose则放入ARGV[1]中

Java调用Lua脚本改造分布式锁

在resources下新建Lua文件：

if(redis.call('get', KEYS[1]) == ARGV[1]) then-- 释放锁return redis.call('del', KEYS[1])
end
return 0

在impl中增加静态变量，防止每次调用unLock函数都要重新调用Lua脚本：

	//DefaultRedisScript是RedisScript的实现类public static final DefaultRedisScript<Long> UNLOCK_SCRIPT;static {UNLOCK_SCRIPT = new DefaultRedisScript<>();UNLOCK_SCRIPT.setLocation(new ClassPathResource("unLock.lua"));//设置脚本位置UNLOCK_SCRIPT.setResultType(Long.class);//配置返回值}

修改unLock函数，调用Lua脚本：

	public void unLock() {//调用Lua脚本stringRedisTemplate.execute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX + name), //转成List类型ID_PREFIX + Thread.currentThread().getId());}

Redisson

基于setnx的分布式锁存在下面的问题：
1、不可重入：同一个线程无法多次获取同一把锁（当同一个线程内，方法A获取了锁，然后调用方法B，方法B中没办法获取同一把锁，就无法执行）
2、不可重试：获取锁只尝试一次就返回false，没有重试机制
3、超时释放：虽然可以避免死锁，但如果业务耗时很长，也会导致锁释放，会再次发生线程安全问题
4、主从一致性问题：若Redis提供了主从集群，主从同步存在延迟，当主宕机时，如果从并同步主中的锁数据，则会出现锁实现
Redisson是一个在Redis基础上实现的分布式工具集合，提供了很多分布式服务，包含了各种分布式锁的实现。

Redisson快速入门

1、引入依赖：

	<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.23.1</version></dependency>

2、配置Redisson客户端：

@Configuration
public class RedissonConfig {@Beanpublic RedissonClient redissonClient(){//配置Config config = new Config();//添加Redis地址，这里添加的是单点的地址，也可以使用config.userClusterServer()来添加集群的地址config.useSingleServer().setAddress("redis://192.168.177.130:6379").setPassword("123456");//创建客户端return Redisson.create(config);}}

3、使用Redisson的分布式锁：
redissonClient注入后，只需要将之前的订单impl的锁的定义换成下面的代码就行了

RLock lock = redissonClient.getLock("lock:order:" + userId);

运行代码，做两个测试：
（1）使用postman发送请求，查看下单是否正常：

（2）jmeter进行多线程测试，测试一人一单功能：

Redisson的可重入锁原理

我们用如下代码片段就可以解决不可重入问题：

//创建锁对象
RLock lock = redissonClient.getLock("lock");
@Test
void method1() {boolean isLock = lock.tryLock();if(!isLock){log.error("获取锁失败，1");return;}try{log.info("获取锁成，1");method2();} finally {log.info("释放锁，1");lock.unlock();}
}
void method2() {boolean isLock = lock.tryLock();if(!isLock){log.error("获取锁失败，2");return;}try{log.info("获取锁成，2");method2();} finally {log.info("释放锁，2");lock.unlock();}
}

可以发现，如果我们使用之前的加锁与释放锁的方法，我们执行method1方法，获取锁成功以后，method1又去执行了method2方法，这时候因为他们是同一个线程，key就是相同的，就会出现method2无法获得锁，导致method2无法执行，从而造成阻塞。
所以，String类型的结构显然就不行了。我们需要找到一种数据结构，能够在一个key里面获取多个东西——Hash：

Hash结构（hset）的KEY对应的VALUE包含了field与value，因此我们可以让KEY对应锁名称，让field对应线程标识，让value位置记录锁的重入次数（初始为0）。

因此，发生上述情况的时候，虽然线程的标识是相同的，但我们可以将重入次数+1，代表第二次获取锁，这时候整体的VALUE是不相同的。
需要注意的是，method2执行完毕以后不能直接释放这个key对应的锁，因为这样的话会导致method1没有执行完毕就被删掉了，解决的方法是让重入次数-1，只有所有业务都执行完了（重入次数=0）的时候才能真正释放。
这样我们的流程就会发生变化（哈希结构没有直接的EX来设置有效期）：

这样的代码就很长了，我们肯定要用Lua脚本来保证代码的原子性，而Lua代码获取锁与释放锁的逻辑已经是保存到RedissonLock类中了，我们只需要直接调用tyrLock与unlock方法就行。
总结：Redisson的可重入原理的核心就是因为我们使用了hash结构，记录了获取锁的线程以及可重用的次数

Redisson的锁重试和WatchDog机制

这里的底层逻辑非常的复杂，都得自己去啃一遍，啃半天都是很有可能的。

Redisson分布式锁原理：
1、可重入：利用hash结构记录线程id和重入次数
2、可重试：利用信号量和PubSub功能实现等待、唤醒，获取锁失败的重试机制
3、超时续约：利用watchDog，每隔一段时间（releaseTime/3），重置超时时间

Redisson的multiLock原理

到此，Redisso解决了不可重入、不可重试、超时释放问题，而主从一致性问题还没解决。
也就是当我们的java对Redis集群的主结点进行获取锁的操作之后，主结点要与从结点保持主从同步，而就在主从同步还未完成的时候，主结点宕机了，需要选出一个从结点来替代成为主结点，但因为主从同步没完成，锁失效了，这样就会发生线程并发问题。

既然产生问题的原因是主从一致，那么就可以考虑不再设置主结点，所有结点一视同仁，获取锁的操作同步对所有的结点进行，并且只有所有的结点都获取锁了，才算获取锁成功。这样即便有结点宕机了也不会产生上述的问题。

当然我们也可以对所有的结点都配备从结点，也就是依旧保持主从同步，也就是说这时候的主结点不再只有一个了，那么主结点宕机后，选出这个主结点的其中一个从结点来替代，也不会发生并发安全问题，因为即便有线程对这台Redis乘虚而入了，也没有办法操作，只有在所有结点都获取锁了，才算成功。

这一套方案就叫做连锁，在这边我配置了3台Redis结点，用于后续测试：

配置很麻烦，但是用Docker就会方便很多，直接在Redis中输入如下命令：

docker pull redis:6.2
docker run -id --name=r1 -p 6380:6379 redis:6.2
docker run -id --name=r2 -p 6381:6379 redis:6.2

创建好以后记得配置Redis是开机自启动的：
Redis：原理速成+项目实战——初识Redis、Redis的安装及启动、Redis客户端

连接的时候要注意端口号分别是6380与6381（我没配置密码，不用填）：

1、先在RedissonConfig中配置好另外2个结点：

2、把三个独立的锁连接在一起，变成连锁：

@Slf4j
@SpringBootTest
public class RedissonTest {@Resourceprivate RedissonClient redissonClient;@Resourceprivate RedissonClient redissonClient2;@Resourceprivate RedissonClient redissonClient3;private RLock lock;@BeforeEachvoid setUp(){RLock lock1 = redissonClient.getLock("order");RLock lock2 = redissonClient2.getLock("order");RLock lock3 = redissonClient3.getLock("order");//创建连锁lock = redissonClient.getMultiLock(lock1, lock2, lock3);}@Testvoid method1() throws InterruptedException {//尝试获取锁boolean isLock = lock.tryLock(1L, TimeUnit.SECONDS);if(!isLock){log.error("获取锁失败，1");return;}try{log.info("获取锁成，1");method2();} finally {log.info("释放锁，1");lock.unlock();}}void method2() {boolean isLock = lock.tryLock();if(!isLock){log.error("获取锁失败，2");return;}try{log.info("获取锁成，2");log.info("开始执行业务2");} finally {log.info("释放锁，2");lock.unlock();}}
}

3、打断点：

debug运行method1，成功获取锁：

可以发现三个Redis都有同一把锁，且value为1：

method2中打断点调试：

value变为2：

unlock，value变回1：

再unlock，锁被释放（不再演示）