关于分布式锁

先别说了别的，先来一个总结。

synchronized 单机版可以，但是上了分布式就不行了。

nginx 分布式服务单机锁就不行

取消单机锁，上redis分布式锁setnx

注意的问题：

如果只加了锁，没有释放锁，出现异常的话。可能无法释放锁，所有必须代码层finally释放锁。

宕机了，部署了微服务代码层根本就没有走到finally这块，没办法保证解锁，这个key没有被删除，需要有lockKey的过期时间设定。

为redis的分布式锁key,增加过期时间，此外，还必须要setnx+过期时间必须在一行，保证原子性。

必须规定只能自己删除自己的锁，你不能把别人的锁删除了，防止张冠李戴，1删2，2删3.

Redis集群环境下，我们自己写的也不ok,就直接上RedLock之Redssion落地实现

redis分布式锁01

*synchronized 是一个关键字

*ReentraLock 是一个类，可以进行异常处理。

class X {private final ReentrantLock lock = new ReentrantLock();// ...public void m() {lock.lock();  // block until condition holds//不见不散try {// ... method body} finally {lock.unlock()}}public void m2() {if(lock.tryLock(timeout, unit)){//过时不候try {// ... method body} finally {lock.unlock()}   }else{// perform alternative actions}}}

redis分布式锁02

分布式部署后，单机锁还是出现超卖现象，需要分布式锁。

redis cluster

Nginx配置负载均衡，Nginx学习笔记or备份

Nginx配置文件修改内容

upstream myserver{server 127.0.0.1:1111;server 127.0.0.1:2222;
}server {listen       80;server_name  localhost;#charset koi8-r;#access_log  logs/host.access.log  main;location / {# 负责用到的配置proxy_pass  http://myserver;root   html;index  index.html index.htm;}#error_page  404              /404.html;# redirect server error pages to the static page /50x.html#error_page   500 502 503 504  /50x.html;location = /50x.html {root   html;}
}

启动两个微服务：1111，2222，多次访问http://localhost/buy_goods，服务提供端口在1111，2222；两者之间横跳。

上边点击，下边模拟高并发

用JMeter模拟测试高并发，100个线程同时访问http://localhost/buy_goods。

启动测试，后台日志：

这就是所谓分布式部署后出现超卖的现象。

Reids具有极高的性能，且具有命令对分布式支持友好，借助SET命令既可以实现枷锁。

EX seconds – Set the specified expire time, in seconds.
PX milliseconds – Set the specified expire time, in milliseconds.
NX – Only set the key if it does not already exist.
XX – Only set the key if it already exist.

在java代码里

public static final String REDIS_LOCK = "redis_lock";@Autowired
private StringRedisTemplate stringRedisTemplate;public void m(){String value = UUID.randomUUID().toString() + Thread.currentThread().getName();Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(REDIS_LOCK, value);if(!flag) {return "抢锁失败";}...//业务逻辑stringRedisTemplate.delete(REDIS_LOCK);
}

Reids分布式锁03

上边Java源码分布式锁的问题：出现异常的话，可能无法释放锁，必须在代码finally释放锁。

解决方法：try…finally…

public static final String REDIS_LOCK = "redis_lock";@Autowired
private StringRedisTemplate stringRedisTemplate;public void m(){String value = UUID.randomUUID().toString() + Thread.currentThread().getName();try{Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(REDIS_LOCK, value);if(!flag) {return "抢锁失败";}...//业务逻辑}finally{stringRedisTemplate.delete(REDIS_LOCK);   }
}

另一个问题：部署了微服务jar包的机器挂了，代码层根本没有走到finally这块，没办法保证释放锁，这个key没有被删除，需要加一个过期时间限定key.

public static final String REDIS_LOCK = "redis_lock";@Autowired
private StringRedisTemplate stringRedisTemplate;public void m(){String value = UUID.randomUUID().toString() + Thread.currentThread().getName();try{Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(REDIS_LOCK, value);//设定时间stringRedisTemplate.expire(REDIS_LOCK, 10L, TimeUnit.SECONDS);if(!flag) {return "抢锁失败";}...//业务逻辑}finally{stringRedisTemplate.delete(REDIS_LOCK);   }
}

Reids分布式锁04

新问题:设置key+过期时间分开了，必须要合并成一行具备原子性。

解决办法：

public static final String REDIS_LOCK = "redis_lock";@Autowired
private StringRedisTemplate stringRedisTemplate;public void m(){String value = UUID.randomUUID().toString() + Thread.currentThread().getName();try{Boolean flag = stringRedisTemplate.opsForValue()//使用另一个带有设置超时操作的方法.setIfAbsent(REDIS_LOCK, value, 10L, TimeUnit.SECONDS);//设定时间//stringRedisTemplate.expire(REDIS_LOCK, 10L, TimeUnit.SECONDS);if(!flag) {return "抢锁失败";}...//业务逻辑}finally{stringRedisTemplate.delete(REDIS_LOCK);   }
}

另一个新的问题：

张冠李戴，删除了别人的锁

解决方法：只能删除自己的，不许动别人的。

public static final String REDIS_LOCK = "redis_lock";@Autowired
private StringRedisTemplate stringRedisTemplate;public void m(){String value = UUID.randomUUID().toString() + Thread.currentThread().getName();try{Boolean flag = stringRedisTemplate.opsForValue()//使用另一个带有设置超时操作的方法.setIfAbsent(REDIS_LOCK, value, 10L, TimeUnit.SECONDS);//设定时间//stringRedisTemplate.expire(REDIS_LOCK, 10L, TimeUnit.SECONDS);if(!flag) {return "抢锁失败";}...//业务逻辑}finally{if(stringRedisTemplate.opsForValue().get(REDIS_LOCK).equals(value)) {stringRedisTemplate.delete(REDIS_LOCK);}}
}

Reids分布式锁05

finally块判断 + del删除操作不是原子性的

用lua脚本

用redis的自身的事务

事务介绍

Redis的事条是通过MULTI，EXEC，DISCARD和WATCH这四个命令来完成。
Redis的单个命令都是原子性的，所以这里确保事务性的对象是命令集合。
Redis将命令集合序列化并确保处于一事务的命令集合连续且不被打断的执行。
Redis不支持回滚的操作。
命令   描述
DISCARD   取消事务，放弃执行事务块内的所有命令。
EXEC   执行所有事务块内的命令。
MULTI   标记一个事务块的开始。
UNWATCH   取消 WATCH 命令对所有 key 的监视。
WATCH key [key …]   监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。

Reids分布式锁06

public static final String REDIS_LOCK = "redis_lock";@Autowired
private StringRedisTemplate stringRedisTemplate;public void m(){String value = UUID.randomUUID().toString() + Thread.currentThread().getName();try{Boolean flag = stringRedisTemplate.opsForValue()//使用另一个带有设置超时操作的方法.setIfAbsent(REDIS_LOCK, value, 10L, TimeUnit.SECONDS);//设定时间//stringRedisTemplate.expire(REDIS_LOCK, 10L, TimeUnit.SECONDS);if(!flag) {return "抢锁失败";}...//业务逻辑}finally{while(true){stringRedisTemplate.watch(REDIS_LOCK);if(stringRedisTemplate.opsForValue().get(REDIS_LOCK).equalsIgnoreCase(value)){stringRedisTemplate.setEnableTransactionSupport(true);stringRedisTemplate.multi();stringRedisTemplate.delete(REDIS_LOCK);List<Object> list = stringRedisTemplate.exec();if (list == null) {continue;}}stringRedisTemplate.unwatch();break;} }
}

Reids分布式锁07

Redis调用Lua脚本通过命令保证代码执行的原子性。

RedisUtils：

import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;public class RedisUtils {private static JedisPool jedisPool;static {JedisPoolConfig jpc = new JedisPoolConfig();jpc.setMaxTotal(20);jpc.setMaxIdle(10);jedisPool = new JedisPool(jpc);}public static JedisPool getJedis() throws Exception{if(jedisPool == null)throw new NullPointerException("JedisPool is not OK.");return jedisPool;}}

public static final String REDIS_LOCK = "redis_lock";@Autowired
private StringRedisTemplate stringRedisTemplate;public void m(){String value = UUID.randomUUID().toString() + Thread.currentThread().getName();try{Boolean flag = stringRedisTemplate.opsForValue()//使用另一个带有设置超时操作的方法.setIfAbsent(REDIS_LOCK, value, 10L, TimeUnit.SECONDS);//设定时间//stringRedisTemplate.expire(REDIS_LOCK, 10L, TimeUnit.SECONDS);if(!flag) {return "抢锁失败";}...//业务逻辑}finally{Jedis jedis = RedisUtils.getJedis();String script = "if redis.call('get', KEYS[1]) == ARGV[1] "+ "then "+ "    return redis.call('del', KEYS[1]) "+ "else "+ "    return 0 "+ "end";try {Object o = jedis.eval(script, Collections.singletonList(REDIS_LOCK),// Collections.singletonList(value));if("1".equals(o.toString())) {System.out.println("---del redis lock ok.");}else {System.out.println("---del redis lock error.");}}finally {if(jedis != null) jedis.close();}}
}

Reids分布式锁08

确保RedisLock过期时间大于业务执行时间的问题

Redis分布式锁如何续期？

集群 + CAP对比ZooKeeper 对比ZooKeeper，重点，CAP

Redis - AP

异步复制造成的锁丢失，比如：主节点没来的及把刚刚set进来这条数据给从节点，就挂了。那么其他的数据就从从机获取数据，就会出现数据的错误。他的选举算法是：主节点获得到key的时，就立即返回说ok了。

ZooKeeper - CP 他的选举机制和Reids不一样，他有他自己的选举算法，zooKeeper获得key的时候先不着急回复，他会先慢慢通知另外两个从节点，保证其他的slave节点和主节点数据一样了，我再返回去跟你说ok了，我们这边加锁成功了。也就是全部成功了，他才会跟你说好了。

注意：就理论而言ZooKeeper比较好，但是一切皆平横，但是他的高可用就会下降。
CAP

C：Consistency（强一致性）
A：Availability（可用性）
P：Partition tolerance（分区容错性）

终于我们推出了在集群的环境下，最牛逼的分布式锁RedLock之Redisson

Reids分布式锁09

Redisson: Redis Java client with features of In-Memory Data Grid

类Redisson配置

import org.redisson.Redisson;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RedisConfig {@Beanpublic Redisson redisson() {Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0);return (Redisson)Redisson.create(config);}}

Redisson模板

public static final String REDIS_LOCK = "REDIS_LOCK";@Autowired
private Redisson redisson;@GetMapping("/doSomething")
public String doSomething(){RLock redissonLock = redisson.getLock(REDIS_LOCK);redissonLock.lock();try {//doSomething}finally {redissonLock.unlock();}
}

回到实例

import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class GoodController{public static final String REDIS_LOCK = "REDIS_LOCK";@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@Autowiredprivate Redisson redisson;@GetMapping("/buy_goods")public String buy_Goods(){//String value = UUID.randomUUID().toString() + Thread.currentThread().getName();RLock redissonLock = redisson.getLock(REDIS_LOCK);redissonLock.lock();try {String result = stringRedisTemplate.opsForValue().get("goods:001");// get key ====看看库存的数量够不够int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0){int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001", String.valueOf(realNumber));System.out.println("成功买到商品，库存还剩下: "+ realNumber + " 件" + "\t服务提供端口" + serverPort);return "成功买到商品，库存还剩下:" + realNumber + " 件" + "\t服务提供端口" + serverPort;}else{System.out.println("商品已经售完/活动结束/调用超时,欢迎下次光临" + "\t服务提供端口" + serverPort);}return "商品已经售完/活动结束/调用超时,欢迎下次光临" + "\t服务提供端口" + serverPort;}finally {redissonLock.unlock();}}}

为了让代码更加严谨在解锁的时候需要注意下：防止出现

IllegalMonitorStateException: attempt to unlock lock，not loked by current thread by node id:da6385f-81a5-4e6c-b8c0

public static final String REDIS_LOCK = "REDIS_LOCK";@Autowired
private Redisson redisson;@GetMapping("/doSomething")
public String doSomething(){RLock redissonLock = redisson.getLock(REDIS_LOCK);redissonLock.lock();try {//doSomething}finally {//添加后，更保险if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()) {redissonLock.unlock();}}
}