以商品超賣為例講解Redis分散式鎖
阿新 • • 發佈：2019-09-21
本案例主要講解
Redis實現分散式鎖的兩種實現方式：Jedis實現、Redisson實現。網上關於這方面講解太多了，Van自認為文筆沒他們好，還是用示例程式碼說明。
一、jedis 實現
該方案只考慮
Redis單機部署的場景
1.1 加鎖
1.1.1 原理
jedis.set(String key, String value, String nxxx, String expx, int time)key: 使用key來當鎖，因為key是唯一的;value: 我傳的是唯一值（UUID），很多童鞋可能不明白，有key作為鎖不就夠了嗎，為什麼還要用到value？原因是分散式鎖要滿足解鈴還須繫鈴人：通過給value賦值為requestId，我們就知道這把鎖是哪個請求加的了，在解鎖的時候要驗證value值，不能誤解鎖;nxxx: 這個引數我填的是NX，意思是SET IF NOT EXIST，即當key不存在時，我們進行set操作；若key已經存在，則不做任何操作；expx: 這個引數我傳的是PX，意思是我們要給這個key加一個過期的設定，具體時間由第五個引數決定;time: 與第四個引數相呼應，代表key的過期時間。
1.1.2 小結
set()加入了NX引數，可以保證如果已有key存在，則函式不會呼叫成功，也就是隻有一個客戶端能持有鎖，滿足互斥性;- 其次，由於我們對鎖設定了過期時間，即使鎖的持有者後續發生崩潰而沒有解鎖，鎖也會因為到了過期時間而自動解鎖（即
key被刪除），不會發生死鎖; - 最後，因為我們將
value賦值為requestId，代表加鎖的客戶端請求標識，那麼在客戶端在解鎖的時候就可以進行校驗是否是同一個客戶端。
1.2 釋放鎖
釋放鎖時需要驗證
value值，也就是說我們在獲取鎖的時候需要設定一個value，不能直接用del key這種粗暴的方式，因為直接del key任何客戶端都可以進行解鎖了，所以解鎖時，我們需要判斷鎖是否是自己的(基於value值來判斷)
- 首先,寫了一個簡單
Lua指令碼程式碼,作用是：獲取鎖對應的value值，檢查是否與requestId相等，如果相等則刪除鎖（解鎖）; - 然後,將
Lua程式碼傳到jedis.eval()方法裡，並使引數KEYS[1]賦值為lockKey，ARGV[1]賦值為requestId。eval()方法是將Lua程式碼交給Redis服務端執行。
1.3 案例（家庭多人領取獎勵的場景）
這裡放出的是關鍵程式碼，詳細可執行的程式碼可至文末地址下載示例程式碼。
1.3.1 準備
該案例模擬家庭內多人通過領取一個獎勵，但是隻能有一個人能領取成功，不能重複領取（之前做過獎勵模組的需求）
family_reward_record表
CREATE TABLE `family_reward_record` (
`id` bigint(10) NOT NULL AUTO_INCREMENT COMMENT '主鍵id',
`family_id` bigint(20) NOT NULL DEFAULT '0' COMMENT '商品名稱',
`reward_type` int(10) NOT NULL DEFAULT '1' COMMENT '商品庫存數量',
`state` int(1) NOT NULL DEFAULT '0' COMMENT '商品狀態',
`create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '入庫時間',
`update_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=270 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='家庭領取獎勵表（家庭內多人只能有一個人能領取成功，不能重複領取）';application.yml
spring:
datasource:
url: jdbc:mysql://47.98.178.84:3306/dev
username: dev
password: password
driver-class-name: com.mysql.jdbc.Driver
redis:
host: 47.98.178.84
port: 6379
password: password
timeout: 2000
# mybatis
mybatis:
mapper-locations: classpath:mapper/*.xml
type-aliases-package: cn.van.mybatis.demo.entity1.3.2 核心實現
- Jedis 單機配置類 -
RedisConfig.java
@Configuration
public class RedisConfig extends CachingConfigurerSupport {
@Value("${spring.redis.host}")
private String host;
@Value("${spring.redis.port}")
private int port;
@Value("${spring.redis.password}")
private String password;
@Value("${spring.redis.timeout}")
private int timeout;
@Bean
public JedisPool redisPoolFactory() {
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
if (StringUtils.isEmpty(password)) {
return new JedisPool(jedisPoolConfig, host, port, timeout);
}
return new JedisPool(jedisPoolConfig, host, port, timeout, password);
}
@Bean(name = "redisTemplate")
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
RedisTemplate<String, Object> redisTemplate = new RedisTemplate();
redisTemplate.setConnectionFactory(redisConnectionFactory);
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.setVisibility(PropertyAccessor.ALL, Visibility.ANY);
objectMapper.enableDefaultTyping(DefaultTyping.NON_FINAL);
Jackson2JsonRedisSerializer<Object> jsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
jsonRedisSerializer.setObjectMapper(objectMapper);
redisTemplate.setDefaultSerializer(jsonRedisSerializer);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.afterPropertiesSet();
return redisTemplate;
}
}- 分散式鎖工具類 -
RedisDistributedLock.java
@Component
public class RedisDistributedLock {
/**
* 成功獲取鎖標示
*/
private static final String LOCK_SUCCESS = "OK";
/**
* 成功解鎖標示
*/
private static final Long RELEASE_SUCCESS = 1L;
@Autowired
private JedisPool jedisPool;
/**
* redis 資料儲存過期時間
*/
final int expireTime = 500;
/**
* 嘗試獲取分散式鎖
* @param lockKey 鎖
* @param lockValue 請求標識
* @return 是否獲取成功
*/
public boolean tryLock(String lockKey, String lockValue) {
Jedis jedis = null;
try{
jedis = jedisPool.getResource();
String result = jedis.set(lockKey, lockValue, "NX", "PX", expireTime);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
} finally {
if(jedis != null){
jedis.close();
}
}
return false;
}
/**
* 釋放分散式鎖
* @param lockKey 鎖
* @param lockValue 請求標識
* @return 是否釋放成功
*/
public boolean unLock(String lockKey, String lockValue) {
Jedis jedis = null;
try {
jedis = jedisPool.getResource();
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(lockValue));
if (RELEASE_SUCCESS.equals(result)) {
return true;
}
} finally {
if(jedis != null){
jedis.close();
}
}
return false;
}
}- 不加鎖時：模擬
familyId = 1的家庭同時領取獎勵
@Override
public HttpResult receiveAward() {
Long familyId = 1L;
Map<String, Object> params = new HashMap<String, Object>(16);
params.put("familyId", familyId);
params.put("rewardType", 1);
int count = familyRewardRecordMapper.selectCountByFamilyIdAndRewardType(params);
if (count == 0) {
FamilyRewardRecordDO recordDO = new FamilyRewardRecordDO(familyId,1,0,LocalDateTime.now());
int num = familyRewardRecordMapper.insert(recordDO);
if (num == 1) {
return HttpResult.success();
}
return HttpResult.failure(-1, "記錄插入失敗");
}
return HttpResult.success("該記錄已存在");
}- 加鎖的實現：模擬
familyId = 2的家庭同時領取獎勵
@Override
public HttpResult receiveAwardLock() {
Long familyId = 2L;
Map<String, Object> params = new HashMap<String, Object>(16);
params.put("familyId", familyId);
params.put("rewardType", 1);
int count = familyRewardRecordMapper.selectCountByFamilyIdAndRewardType(params);
if (count == 0) {
// 沒有記錄則建立領取記錄
FamilyRewardRecordDO recordDO = new FamilyRewardRecordDO(familyId,1,0,LocalDateTime.now());
// 分散式鎖的key(familyId + rewardType)
String lockKey = recordDO.getFamilyId() + "_" + recordDO.getRewardType();
// 分散式鎖的value(唯一值)
String lockValue = createUUID();
boolean lockStatus = redisLock.tryLock(lockKey, lockValue);
// 鎖被佔用
if (!lockStatus) {
log.info("鎖已經佔用了");
return HttpResult.failure(-1,"失敗");
}
// 不管多個請求，加鎖之後，只會有一個請求能拿到鎖，進行插入操作
log.info("拿到了鎖，當前時刻:{}",System.currentTimeMillis());
int num = familyRewardRecordMapper.insert(recordDO);
if (num != 1) {
log.info("資料插入失敗！");
return HttpResult.failure(-1, "資料插入失敗！");
}
log.info("資料插入成功！準備解鎖...");
boolean unLockState = redisLock.unLock(lockKey,lockValue);
if (!unLockState) {
log.info("解鎖失敗！");
return HttpResult.failure(-1, "解鎖失敗！");
}
log.info("解鎖成功！");
return HttpResult.success();
}
log.info("該記錄已存在");
return HttpResult.success("該記錄已存在");
}
private String createUUID() {
UUID uuid = UUID.randomUUID();
String str = uuid.toString().replace("-", "_");
return str;
}1.3.3 測試
我採用的是
JMeter工具進行測試，加鎖和不加鎖的情況都設定成：五次併發請求。
1.3.3.1 不加鎖
/**
* 家庭成員領取獎勵（不加鎖）
* @return
*/
@PostMapping("/receiveAward")
public HttpResult receiveAward() {
return redisLockService.receiveAward();
}- 請求方式：
POST - 請求地址：http://localhost:8080/redisLock/receiveAward
- 返回結果：插入了五條記錄
1.3.3.2 加鎖
/**
* 家庭成員領取獎勵（加鎖）
* @return
*/
@PostMapping("/receiveAwardLock")
public HttpResult receiveAwardLock() {
return redisLockService.receiveAwardLock();
}- 請求方式：
POST - 請求地址：http://localhost:8080/redisLock/receiveAwardLock
- 返回結果：只插入了一條記錄
通過對比，說明分散式鎖起作用了。
1.4 小結
我上家使用的就是這種加鎖方式，看上去很OK，實際上在Redis叢集的時候會出現問題，比如：
A客戶端在Redis的master節點上拿到了鎖，但是這個加鎖的key還沒有同步到slave節點，master故障，發生故障轉移，一個slave節點升級為master節點，B客戶端也可以獲取同個key的鎖，但客戶端A也已經拿到鎖了，這就導致多個客戶端都拿到鎖。
正因為如此，Redis作者antirez基於分散式環境下提出了一種更高階的分散式鎖的實現方式：Redlock。
二、Redlock實現
2.1 原理
antirez提出的Redlock演算法大概是這樣的：
在Redis的分散式環境中，我們假設有N個Redis master。這些節點完全互相獨立，不存在主從複製或者其他叢集協調機制。我們確保將在N個例項上使用與在Redis單例項下相同方法獲取和釋放鎖。現在我們假設有5個Redis master節點，同時我們需要在5臺伺服器上面執行這些Redis例項，這樣保證他們不會同時都宕掉。
2.1.1 加鎖
為了取到鎖，客戶端應該執行以下操作(RedLock演算法加鎖步驟):
- 獲取當前
Unix時間，以毫秒為單位; - 依次嘗試從
5個例項，使用相同的key和具有唯一性的value（例如UUID）獲取鎖。當向Redis請求獲取鎖時，客戶端應該設定一個網路連線和響應超時時間，這個超時時間應該小於鎖的失效時間。例如你的鎖自動失效時間為10秒，則超時時間應該在5-50毫秒之間。這樣可以避免伺服器端Redis已經掛掉的情況下，客戶端還在死死地等待響應結果。如果伺服器端沒有在規定時間內響應，客戶端應該儘快嘗試去另外一個Redis例項請求獲取鎖; - 客戶端使用當前時間減去開始獲取鎖時間（步驟
1記錄的時間）就得到獲取鎖使用的時間。當且僅當從大多數（N/2+1，這裡是3個節點）的Redis節點都取到鎖，並且使用的時間小於鎖失效時間時，鎖才算獲取成功; - 如果取到了鎖，
key的真正有效時間等於有效時間減去獲取鎖所使用的時間（步驟3計算的結果）。 - 如果因為某些原因，獲取鎖失敗（沒有在至少
N/2+1個Redis例項取到鎖或者取鎖時間已經超過了有效時間），客戶端應該在所有的Redis例項上進行解鎖（即便某些Redis例項根本就沒有加鎖成功，防止某些節點獲取到鎖但是客戶端沒有得到響應而導致接下來的一段時間不能被重新獲取鎖）。
2.1.2 解鎖
向所有的Redis例項傳送釋放鎖命令即可，不用關心之前有沒有從Redis例項成功獲取到鎖.
2.2 案例（商品超賣為例）
這部分以最常見的案例：搶購時的商品超賣（庫存數減少為負數）為例
2.2.1 準備
good表
CREATE TABLE `good` (
`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主鍵id',
`good_name` varchar(255) NOT NULL COMMENT '商品名稱',
`good_counts` int(255) NOT NULL COMMENT '商品庫存',
`create_time` timestamp NOT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '建立時間',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COMMENT='商品表';
-- 插入兩條測試資料
INSERT INTO `good` VALUES (1, '哇哈哈', 5, '2019-09-20 17:39:04');
INSERT INTO `good` VALUES (2, '衛龍', 5, '2019-09-20 17:39:06');- 配置檔案跟上面一樣
2.2.2 核心實現
Redisson配置類RedissonConfig.java
我這裡配置的是單機，更多配置詳見https://github.com/redisson/redisson/wiki/配置
@Configuration
public class RedissonConfig {
@Value("${spring.redis.host}")
private String host;
@Value("${spring.redis.port}")
private String port;
@Value("${spring.redis.password}")
private String password;
/**
* RedissonClient,單機模式
* @return
* @throws IOException
*/
@Bean
public RedissonClient redissonSentinel() {
//支援單機，主從，哨兵，叢集等模式,此為單機模式
Config config = new Config();
config.useSingleServer()
.setAddress("redis://" + host + ":" + port)
.setPassword(password);
return Redisson.create(config);
}
}- 不加鎖時
@Override
public HttpResult saleGoods(){
// 以指定goodId = 1：哇哈哈為例
Long goodId = 1L;
GoodDO goodDO = goodMapper.selectByPrimaryKey(goodId);
int goodStock = goodDO.getGoodCounts();
if (goodStock >= 1) {
goodMapper.saleOneGood(goodId);
}
return HttpResult.success();
}- 加鎖
@Override
public HttpResult saleGoodsLock(){
// 以指定goodId = 2：衛龍為例
Long goodId = 2L;
GoodDO goodDO = goodMapper.selectByPrimaryKey(goodId);
int goodStock = goodDO.getGoodCounts();
String key = goodDO.getGoodName();
log.info("{}剩餘總庫存,{}件", key,goodStock);
// 將商品的實時庫存放在redis 中，便於讀取
stringRedisTemplate.opsForValue().set(key, Integer.toString(goodStock));
// redisson 鎖 的key
String lockKey = goodDO.getId() +"_" + key;
RLock lock = redissonClient.getLock(lockKey);
// 設定60秒自動釋放鎖 （預設是30秒自動過期）
lock.lock(60, TimeUnit.SECONDS);
// 此步開始，序列銷售
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
// 如果快取中庫存量大於1，可以繼續銷售
if (stock >= 1) {
goodDO.setGoodCounts(stock - 1);
int num = goodMapper.saleOneGood(goodId);
if (num == 1) {
// 減庫存成功，將快取同步
stringRedisTemplate.opsForValue().set(key,Integer.toString((stock-1)));
}
log.info("{},當前庫存,{}件", key,stock);
}
lock.unlock();
return HttpResult.success();
}2.3 測試
採用的是
JMeter工具進行測試，初始化的時候兩個商品的庫存設定都是：5;所以這裡加鎖和不加鎖的情況都設定成：十次併發請求。
2.3.1 不加鎖
/**
* 售賣商品(不加鎖)
* @return
*/
@PostMapping("/saleGoods")
public HttpResult saleGoods() {
return redisLockService.saleGoods();
}- 請求方式：
POST - 請求地址：http://localhost:8080/redisLock/saleGoods
- 返回結果：
id =1的商品庫存減為-5
2.3.2 加鎖
/**
* 售賣商品(加鎖)
* @return
*/
@PostMapping("/saleGoodsLock")
public HttpResult saleGoodsLock() {
return redisLockService.saleGoodsLock();
}- 請求方式：
POST - 請求地址：http://localhost:8080/redisLock/saleGoodsLock
- 返回結果：
id =1的商品庫存減為0
2.3.3 小結
通過2.3.1和2.3.2的結果對比很明顯：前者出現了超賣情況，庫存數賣到了-5，這是決不允許的；而加了鎖的情況後，庫存只會減少到0，便不再銷售。
三、總結
再次說明：以上程式碼不全，如需嘗試，請前往Van 的 Github 檢視完整示例程式碼
第一種基於Redis的分散式鎖並不適合用於生產環境。Redisson 可用於生產環境。當然，分散式的選擇還有Zookeeper的選項，Van後續會整理出來供大家參考。
3.1 示例原始碼地址
https://github.com/vanDusty/SpringBoot-Home/tree/master/springboot-demo-lock/redis-lock
3.2 技術交流
- 風塵部落格
- 風塵部落格-掘金
- 風塵部落格-CSDN