隨著現(xiàn)在分布式架構(gòu)越來越盛行，在很多場景下需要使用到分布式鎖。很多小伙伴對于分布式鎖還不是特別了解，所以特地總結(jié)了一篇文章，讓大家一文讀懂分布式鎖的前世今生。
分布式鎖的實(shí)現(xiàn)有很多種，比如基于數(shù)據(jù)庫、Redis 、 zookeeper 等實(shí)現(xiàn)，本文的示例主要介紹使用Redis實(shí)現(xiàn)分布式鎖。
一、什么是分布式鎖
分布式鎖，即分布式系統(tǒng)中的鎖，分布式鎖是控制分布式系統(tǒng)有序地對共享資源進(jìn)行操作，在單體應(yīng)用中我們通過鎖實(shí)現(xiàn)共享資源訪問，而分布式鎖，就是解決了分布式系統(tǒng)中控制共享資源訪問的問題。
可能初學(xué)的小伙伴就會有疑問，Java多線程中的公平鎖、非公平鎖、自旋鎖、可重入鎖、讀寫鎖、互斥鎖這些都還沒鬧明白呢？怎么有出來一個(gè)分布式鎖？
其實(shí)，可以這么理解：Java的原生鎖是解決多線程下對于共享資源的操作，而分布式鎖則是多進(jìn)程下對于共享資源的操作。分布式系統(tǒng)中競爭共享資源的最小粒度從線程升級成了進(jìn)程。
分布式鎖已經(jīng)被應(yīng)用到各種高并發(fā)的場景下，典型場景案例包括：秒殺、車票、訂單、退款、庫存等場景。
二、為什么要使用分布式鎖
在分布式系統(tǒng)中，常常需要協(xié)調(diào)他們的動作。如果不同的系統(tǒng)或是同一個(gè)系統(tǒng)的不同主機(jī)之間共享了一個(gè)或一組資源，那么訪問這些資源的時(shí)候，往往需要互斥來防止彼此干擾來保證一致性，這個(gè)時(shí)候，便需要使用到分布式鎖。
目前幾乎很多大型網(wǎng)站及應(yīng)用都是分布式部署的，如何保證分布式場景中的數(shù)據(jù)一致性問題一直是一個(gè)比較重要的話題。在某些場景下，為了保證數(shù)據(jù)的完整性和一致性，我們需要保證一個(gè)方法在同一時(shí)間內(nèi)只能被同一個(gè)線程執(zhí)行，這就需要使用分布式鎖。
如上圖所示，假設(shè)用戶A和用戶B同時(shí)購買了某款商品，訂單創(chuàng)建成功后，下單系統(tǒng)A和下單系統(tǒng)B就會同時(shí)對數(shù)據(jù)庫中的該款商品的庫存進(jìn)行扣減。如果此時(shí)不加任何控制，系統(tǒng)B提交的數(shù)據(jù)更新就會覆蓋系統(tǒng)A的數(shù)據(jù)，導(dǎo)致庫存錯(cuò)誤，超賣等問題。
三、分布式鎖應(yīng)該具備哪些條件
在介紹分布式鎖的實(shí)現(xiàn)方式之前，先了解一下分布式鎖應(yīng)該具備哪些條件：
1、在分布式系統(tǒng)環(huán)境下，一個(gè)方法在同一時(shí)間只能被一個(gè)機(jī)器的一個(gè)線程執(zhí)行；
2、高可用的獲取鎖與釋放鎖；
3、高性能的獲取鎖與釋放鎖；
4、具備可重入特性；
5、具備鎖失效機(jī)制，防止死鎖；
6、具備非阻塞鎖特性，即沒有獲取到鎖將直接返回獲取鎖失敗。
四、分布式鎖的實(shí)現(xiàn)方式
隨著業(yè)務(wù)發(fā)展的需要，原來的單體應(yīng)用被演化成分布式集群系統(tǒng)后，由于系統(tǒng)分布在不同機(jī)器上，這就使得原有的并發(fā)控制鎖策略失效，為了解決這個(gè)問題就需要一種跨進(jìn)程的互斥機(jī)制來控制共享資源的訪問，這就需要用到分布式鎖！
分布式鎖的實(shí)現(xiàn)有多種方式，下面介紹下這幾種分布式鎖的實(shí)現(xiàn)方式：
- 基于數(shù)據(jù)庫實(shí)現(xiàn)分布式鎖，(適用于并發(fā)小的系統(tǒng))；
- 基于緩存（Redis等）實(shí)現(xiàn)分布式鎖，（效率高，最流行，存在鎖超時(shí)的問題）；
- 基于Zookeeper實(shí)現(xiàn)分布式鎖，（可靠，但是效率不高）；
盡管有這三種方案，但是不同的業(yè)務(wù)也要根據(jù)自己的情況進(jìn)行選型，他們之間沒有最好只有更適合！
五、基于Redis實(shí)現(xiàn)分布式鎖
使用Redis實(shí)現(xiàn)分布式鎖是目前比較流行的解決方案，主要是使用Redis 獲取鎖與釋放鎖效率都很高，實(shí)現(xiàn)方式也特別簡單。
實(shí)現(xiàn)原理：
（1）獲取鎖的時(shí)候，使用setnx加鎖，并使用expire命令為鎖添加一個(gè)超時(shí)時(shí)間，超過該時(shí)間則自動釋放鎖，鎖的value值為當(dāng)前線程ID，通過此在釋放鎖的時(shí)候進(jìn)行判斷。
（2）獲取鎖的時(shí)候還設(shè)置一個(gè)獲取的超時(shí)時(shí)間，若超過這個(gè)時(shí)間則放棄獲取鎖。
（3）釋放鎖的時(shí)候，通過線程ID判斷是不是該鎖，若是該鎖，則執(zhí)行delete進(jìn)行鎖釋放。
接下來我們就一步一步實(shí)現(xiàn)Redis 分布式鎖。
第一步，創(chuàng)建Spring Boot項(xiàng)目，并引入相關(guān)依賴。
org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-data-redis org.springframework.boot spring-boot-starter org.springframework.boot spring-boot-starter-test test org.apache.commons commons-lang3 com.alibaba fastjson 1.2.72 org.projectlombok lombok
第二步，創(chuàng)建Redis分布式鎖通用操作類，示例代碼如下：
import com.alibaba.fastjson.JSON;import lombok.extern.slf4j.Slf4j;import org.apache.commons.lang3.StringUtils;import org.springframework.context.annotation.Bean;import org.springframework.data.redis.core.RedisTemplate;import org.springframework.data.redis.core.script.DefaultRedisScript;import org.springframework.scheduling.annotation.Async;import org.springframework.scheduling.annotation.Scheduled;import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import org.springframework.stereotype.Component;import javax.annotation.Resource;import java.util.ArrayList;import java.util.List;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;import java.util.concurrent.Executor;import java.util.concurrent.ThreadPoolExecutor;import java.util.concurrent.TimeUnit;/** * Lua腳本 * // 加鎖 * if * redis.call(‘setNx’,KEYS[1],ARGV[1]) * then * if redis.call(‘get’,KEYS[1])==ARGV[1] * return redis.call(‘expire’,KEYS[1],ARGV[2]) * else * return 0 * end * end * * // 解鎖 * redis.call(‘get’, KEYS[1]) == ARGV[1] * then * return redis.call(‘del’, KEYS[1]) * else * return 0 * * * //更新時(shí)間 * if redis.call(‘get’, KEYS[1]) == ARGV[1] then return redis.call(‘expire’, KEYS[1], ARGV[2]) else return 0 end * * * * */@Slf4j@Componentpublic class RedisLockUtils { @Resource private RedisTemplate redisTemplate; private static Map lockInfoMap = new ConcurrentHashMap(); private static final Long SUCCESS = 1L; @Data public static class LockInfo { private String key; private String value; private int expireTime; //更新時(shí)間 private long renewalTime; //更新間隔 private long renewalInterval; public static LockInfo getLockInfo(String key, String value, int expireTime) { LockInfo lockInfo = new LockInfo(); lockInfo.setKey(key); lockInfo.setValue(value); lockInfo.setExpireTime(expireTime); lockInfo.setRenewalTime(System.currentTimeMillis()); lockInfo.setRenewalInterval(expireTime * 2000 / 3); return lockInfo; } } /** * 使用lua腳本加鎖 * @param lockKey 鎖 * @param value 身份標(biāo)識（保證鎖不會被其他人釋放） * @param expireTime 鎖的過期時(shí)間（單位：秒） * @Desc 注意事項(xiàng)，redisConfig配置里面必須使用 genericToStringSerializer序列化,否則獲取不了返回值 */ public boolean tryLock(String lockKey, String value, int expireTime) { String luaScript = “if redis.call(‘setNx’,KEYS[1],ARGV[1]) then if redis.call(‘get’,KEYS[1])==ARGV[1] then return redis.call(‘expire’,KEYS[1],ARGV[2]) else return 0 end end”; DefaultRedisScript redisScript = new DefaultRedisScript(); redisScript.setResultType(Boolean.class); redisScript.setScriptText(luaScript); List keys = new ArrayList(); keys.add(lockKey); //Object result = redisTemplate.execute(redisScript, Collections.singletonList(lockKey),value,expireTime + “”); // Object result = redisTemplate.execute(redisScript, new StringRedisSerializer(), new StringRedisSerializer(), Collections.singletonList(lockKey), identity, expireTime); Object result = redisTemplate.execute(redisScript, keys, value, expireTime); log.info(“已獲取到{}對應(yīng)的鎖!”, lockKey); if (expireTime >= 10) { lockInfoMap.put(lockKey + value, LockInfo.getLockInfo(lockKey, value, expireTime)); } return (boolean) result; } /** * 使用lua腳本釋放鎖 * @param lockKey * @param value * @return 成功返回true, 失敗返回false */ public boolean unlock(String lockKey, String value) { lockInfoMap.remove(lockKey + value); String luaScript = “if redis.call(‘get’, KEYS[1]) == ARGV[1] then return redis.call(‘del’, KEYS[1]) else return 0 end”; DefaultRedisScript redisScript = new DefaultRedisScript(); redisScript.setResultType(Boolean.class); redisScript.setScriptText(luaScript); List keys = new ArrayList(); keys.add(lockKey); Object result = redisTemplate.execute(redisScript, keys, value); log.info(“解鎖成功：{}”, result); return (boolean) result; } /** * 使用lua腳本更新redis鎖的過期時(shí)間 * @param lockKey * @param value * @return 成功返回true, 失敗返回false */ public boolean renewal(String lockKey, String value, int expireTime) { String luaScript = “if redis.call(‘get’, KEYS[1]) == ARGV[1] then return redis.call(‘expire’, KEYS[1], ARGV[2]) else return 0 end”; DefaultRedisScript redisScript = new DefaultRedisScript(); redisScript.setResultType(Boolean.class); redisScript.setScriptText(luaScript); List keys = new ArrayList(); keys.add(lockKey); Object result = redisTemplate.execute(redisScript, keys, value, expireTime); log.info(“更新redis鎖的過期時(shí)間：{}”, result); return (boolean) result; } /** * * @param lockKey 鎖 * @param value 身份標(biāo)識（保證鎖不會被其他人釋放） * @param expireTime 鎖的過期時(shí)間（單位：秒） * @return 成功返回true, 失敗返回false */ public boolean lock(String lockKey, String value, long expireTime) { return redisTemplate.opsForValue().setIfAbsent(lockKey, value, expireTime, TimeUnit.SECONDS); } /** * redisTemplate解鎖 * @param key * @param value * @return 成功返回true, 失敗返回false */ public boolean unlock2(String key, String value) { Object currentValue = redisTemplate.opsForValue().get(key); boolean result = false; if (StringUtils.isNotEmpty(String.valueOf(currentValue)) && currentValue.equals(value)) { result = redisTemplate.opsForValue().getOperations().delete(key); } return result; } /** * 定時(shí)去檢查redis鎖的過期時(shí)間 */ @Scheduled(fixedRate = 5000L) @Async(“redisExecutor”) public void renewal() { long now = System.currentTimeMillis(); for (Map.Entry lockInfoEntry : lockInfoMap.entrySet()) { LockInfo lockInfo = lockInfoEntry.getValue(); if (lockInfo.getRenewalTime() + lockInfo.getRenewalInterval() < now) { renewal(lockInfo.getKey(), lockInfo.getValue(), lockInfo.getExpireTime()); lockInfo.setRenewalTime(now); log.info("lockInfo {}", JSON.toJSONString(lockInfo)); } } } /** * 分布式鎖設(shè)置單獨(dú)線程池 * @return */ @Bean("redisExecutor") public Executor redisExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(1); executor.setMaxPoolSize(1); executor.setQueueCapacity(1); executor.setKeepAliveSeconds(60); executor.setThreadNamePrefix("redis-renewal-"); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy()); return executor; }}
第三步，創(chuàng)建RedisTemplate 配置類，配置Redistemplate,示例代碼如下：
@Configurationpublic class RedisConfig { @Bean public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) throws Exception { RedisTemplate redisTemplate = new RedisTemplate(); redisTemplate.setConnectionFactory(redisConnectionFactory); // 創(chuàng)建 序列化類 GenericToStringSerializer genericToStringSerializer = new GenericToStringSerializer(Object.class); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(genericToStringSerializer); return redisTemplate; }}
第四步，實(shí)現(xiàn)業(yè)務(wù)調(diào)用，這里以扣減庫存為例，示例代碼如下：
@RestControllerpublic class IndexController { @Resource private RedisTemplate redisTemplate; @Autowired private RedisLockUtils redisLock; @RequestMapping(“/deduct-stock”) public String deductStock() { String productId = “product001”; System.out.println(“—————->>>開始扣減庫存”); String key = productId; String requestId = productId + Thread.currentThread().getId(); try { boolean locked = redisLock.lock(key, requestId, 10); if (!locked) { return “error”; } //執(zhí)行業(yè)務(wù)邏輯 //System.out.println(“—————->>>執(zhí)行業(yè)務(wù)邏輯:”+appTitle); int stock = Integer.parseInt(redisTemplate.opsForValue().get(“product001-stock”).toString()); int currentStock = stock-1; redisTemplate.opsForValue().set(“product001-stock”,currentStock); try { Random random = new Random(); Thread.sleep(random.nextInt(3) *1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(“—————->>>扣減庫存結(jié)束:current stock:” + currentStock); return “success,current stock:” + currentStock; } finally { redisLock.unlock2(key, requestId); } }}
六、驗(yàn)證測試
代碼完成之后，開始測試。我們同時(shí)啟動兩個(gè)實(shí)例，端口號為：8888和8889模擬分布式系統(tǒng)。
接下來，我們分別請求：http://localhost:8888/deduct-stock和http://localhost:8889/deduct-stock，或者使用JMater分別請求這兩個(gè)地址，模擬高并發(fā)的情況。
通過上圖我們可以看到，在批量請求的情況下，庫存扣減也沒有出現(xiàn)問題。說明分布式鎖生效了。
最后
以上，我們就把什么是分布式鎖，如何基于Redis 實(shí)現(xiàn)分布式鎖的解決方案介紹完了。分布式鎖是分布式系統(tǒng)中的重要功能組件，希望大家能夠熟練掌握。
