分布式锁,是指在分布式的集群环境中,保证不同节点的线程同步执行。
分布式锁的实现有哪些?
1.Memcached分布式锁
利用Memcached的add命令。此命令是原子性操作,只有在key不存在的情况下,才能add成功,也就意味着线程得到了锁。
2.Redis分布式锁
和Memcached的方式类似,利用Redis的setnx命令。此命令同样是原子性操作,只有在key不存在的情况下,才能set成功。(setnx命令并不完善,后续会介绍替代方案)
3.Zookeeper分布式锁
利用Zookeeper的顺序临时节点,来实现分布式锁和等待队列。Zookeeper设计的初衷,就是为了实现分布式锁服务的。
4.Chubby
Google公司实现的粗粒度分布式锁服务,底层利用了Paxos一致性算法。
这么多种实现方法,选择比较有代表性的Redis的分布式锁来学习:
如何用Redis实现分布式锁?
Redis分布式锁的基本流程并不难理解,但要想写得尽善尽美,也并不是那么容易。在这里,我们需要先了解分布式锁实现的三个核心要素:
1.加锁
最简单的方法是使用setnx命令。key是锁的唯一标识,按业务来决定命名。比如想要给一种商品的秒杀活动加锁,可以给key命名为 “lock_sale_商品ID” 。而value设置成什么呢?我们可以姑且设置成1。加锁的伪代码如下:
setnx(key,1)
当一个线程执行setnx返回1,说明key原本不存在,该线程成功得到了锁;当一个线程执行setnx返回0,说明key已经存在,该线程抢锁失败。
2.解锁
有加锁就得有解锁。当得到锁的线程执行完任务,需要释放锁,以便其他线程可以进入。释放锁的最简单方式是执行del指令,伪代码如下:
del(key)
释放锁之后,其他线程就可以继续执行setnx命令来获得锁。
3.锁超时
锁超时是什么意思呢?如果一个得到锁的线程在执行任务的过程中挂掉,来不及显式地释放锁,这块资源将会永远被锁住,别的线程再也别想进来。
所以,setnx的key必须设置一个超时时间,以保证即使没有被显式释放,这把锁也要在一定时间后自动释放。setnx不支持超时参数,所以需要额外的指令,伪代码如下:
expire(key, 30)
模拟此场景,写一个抢购秒杀的demo:
Controller:
@RestController @RequestMapping("/skill") @Slf4j public class SecKillController { @Autowired private SecKillService secKillService;
/** * 秒杀,没有抢到获得"哎呦喂,xxxxx",抢到了会返回剩余的库存量 * @param productId * @return * @throws Exception */ @GetMapping("/order/{productId}") public String skill(@PathVariable String productId)throws Exception { log.info("@skill request, productId:" + productId); secKillService.orderProductMockDiffUser(productId); return secKillService.querySecKillProductInfo(productId); } }
业务层Impl:(未做任何同步处理)
@Service public class SecKillServiceImpl implements SecKillService { private static final int TIMEOUT = 10 * 1000; //超时时间 10s /** * 国庆活动,皮蛋粥特价,限量100000份 */ static Map<String,Integer> products; static Map<String,Integer> stock; static Map<String,String> orders; static { /** * 模拟多个表,商品信息表,库存表,秒杀成功订单表 */ products = new HashMap<>(); stock = new HashMap<>(); orders = new HashMap<>(); products.put("123456", 100000); stock.put("123456", 100000); } private String queryMap(String productId) { return "国庆活动,皮蛋粥特价,限量份" + products.get(productId) +" 还剩:" + stock.get(productId)+" 份" +" 该商品成功下单用户数目:" + orders.size() +" 人" ; } @Override public String querySecKillProductInfo(String productId) { return this.queryMap(productId); } @Override public void orderProductMockDiffUser(String productId) { Long time = System.currentTimeMillis() + TIMEOUT; //1.查询该商品库存,为0则活动结束。 int stockNum = stock.get(productId); if(stockNum == 0) { throw new SellException(100,"活动结束"); }else { //2.下单(模拟不同用户openid不同) orders.put(KeyUtil.getUniqueKey(),productId); //3.减库存 stockNum =stockNum-1; try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } stock.put(productId,stockNum); } } }
启动项目,然后使用apache bench 压测:ab -n 100 -c 100 http://localhost:8080/skill/order/123456
发现数据同步失败:
接下来尝试在函数加上 synchronized,同步没问题,但是响应时间较长
使用Redis分布式锁:(需要引入 spring-boot-starter-data-redis 相关依赖)
RedisLock类:
@Component @Slf4j public class RedisLock { @Autowired private StringRedisTemplate redisTemplate; /** * 加锁 * @param key * @param value * @return */ public boolean lock(String key, String value){ // 设置redis值,如果值已存在不做操作,跳到下一步 if (redisTemplate.opsForValue().setIfAbsent(key, value)) { return true; } // 获取reids中的时间戳 String currentValue = redisTemplate.opsForValue().get(key); if (!StringUtils.isEmpty(currentValue) && Long.parseLong(currentValue) < System.currentTimeMillis()) { // 拿到上一次的时间戳,并设置新的时间戳,保证只有一个线程能同步 String oldValue = redisTemplate.opsForValue().getAndSet(key, value); // 若第二个线程进来,此时oldvalue已经不等于currentValue了 if (!StringUtils.isEmpty(oldValue) && oldValue.equals(currentValue)) { return true; } } return false; } /** * 解锁 * @param key * @param value */ public void unLock(String key, String value) { try { String currentValue = redisTemplate.opsForValue().get(key); if (!StringUtils.isEmpty(currentValue) && value.equals(currentValue)) { redisTemplate.opsForValue().getOperations().delete(key); } } catch (Exception e) { log.error("【redis分布式锁】解锁异常, {}", e); } } }
业务层Impl:(加上Redis锁的处理)
@Service public class SecKillServiceImpl implements SecKillService { private static final int TIMEOUT = 10 * 1000; //超时时间 10s @Autowired private RedisLock redisLock; /** * 国庆活动,皮蛋粥特价,限量100000份 */ static Map<String,Integer> products; static Map<String,Integer> stock; static Map<String,String> orders; static { /** * 模拟多个表,商品信息表,库存表,秒杀成功订单表 */ products = new HashMap<>(); stock = new HashMap<>(); orders = new HashMap<>(); products.put("123456", 100000); stock.put("123456", 100000); } private String queryMap(String productId) { return "国庆活动,皮蛋粥特价,限量份" + products.get(productId) +" 还剩:" + stock.get(productId)+" 份" +" 该商品成功下单用户数目:" + orders.size() +" 人" ; } @Override public String querySecKillProductInfo(String productId) { return this.queryMap(productId); } @Override public void orderProductMockDiffUser(String productId) { Long time = System.currentTimeMillis() + TIMEOUT; //加锁 if (!redisLock.lock(productId, String.valueOf(time))) { throw new SellException(101, "人太多了歇一会吧!"); } //1.查询该商品库存,为0则活动结束。 int stockNum = stock.get(productId); if(stockNum == 0) { throw new SellException(100,"活动结束"); }else { //2.下单(模拟不同用户openid不同) orders.put(KeyUtil.getUniqueKey(),productId); //3.减库存 stockNum =stockNum-1; try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } stock.put(productId,stockNum); } //解锁 redisLock.unLock(productId, String.valueOf(time)); } }
重启,再次用apache bench压测 ab -n 100 -c 100 http://localhost:8080/skill/order/123456
结果,响应时间非常快,减少了卡顿,同步也正常!