Redis分散式鎖的使用與實現原理
阿新 • • 發佈：2020-11-21
模擬一個電商裡面下單減庫存的場景。
1.首先在redis里加入商品庫存數量。
2.新建一個Spring Boot專案，在pom裡面引入相關的依賴。
```
org.springframework.boot
spring-boot-starter-web
org.springframework.boot
spring-boot-starter-data-redis
```
3.接下來，在application.yml配置redis屬性和指定應用的埠號：
```
server:
port: 8090
spring:
redis:
host: 192.168.0.60
port: 6379
```
4.新建一個Controller類，扣減庫存第一版程式碼：
```
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.Objects;
@RestController
public class StockController {
private static final Logger logger = LoggerFactory.getLogger(StockController.class);
@Resource
private StringRedisTemplate stringRedisTemplate;
@RequestMapping("/reduceStock")
public String reduceStock() {
// 從redis中獲取庫存數量
int stock = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get("stockCount")));
if (stock >
org.redisson
redisson
3.6.5
```
配置類：
```
@Configuration
public class RedissonConfig {
@Bean
public Redisson redisson() {
// 單機模式
Config config = new Config();
config.useSingleServer().setAddress("redis://192.168.0.60:6379").setDatabase(0);
return (Redisson) Redisson.create(config);
}
}
```
接下來用redisson重寫上面的減庫存操作：
```
@Resource
private Redisson redisson;
@RequestMapping("/reduceStock")
public String reduceStock() {
String lockKey = "stockKey";
RLock redissonLock = redisson.getLock(lockKey);
try {
// 加鎖，鎖續命
redissonLock.lock();
// 從redis中獲取庫存數量
int stock = Integer.parseInt(Objects.requireNonNull(stringRedisTemplate.opsForValue().get("stockCount")));
if (stock > 0) {
// 減庫存
int restStock = stock - 1;
// 剩餘庫存再重新設定到redis中
stringRedisTemplate.opsForValue().set("stockCount", String.valueOf(restStock));
logger.info("扣減成功，剩餘庫存：{}", restStock);
} else {
logger.info("庫存不足，扣減失敗。");
}
} finally {
redissonLock.unlock();
}
return "success";
}
```
其實就是三個步驟：獲取鎖，加鎖，釋放鎖。
先簡單看下Redisson的實現原理：
這裡先說一下Redis很多操作使用Lua指令碼來實現原子性操作，關於Lua語法，可以去網上找下相關教程。
使用Lua指令碼的好處有：
1.減少網路開銷，多個命令可以使用一次請求完成；
2.實現了原子性操作，Redis會把Lua指令碼作為一個整體去執行；
3.實現事務，Redis自帶的事務功能有限，而Lua指令碼實現了事務的常規操作，而且還支援回滾。
但是Lua實際上不會使用很多，如果Lua指令碼執行時間過長，因為Redis是單執行緒，因此會導致堵塞。
最後，說下Redisson分散式鎖的程式碼實現，
找到上面的redissonLock.lock()；
lock方法點進去，一直點到RedissonLock類裡面的lockInterruptibly方法：
```
@Override
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
// 獲取執行緒id
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return;
}
RFuture future = subscribe(threadId);
commandExecutor.syncSubscription(future);
try {
while (true) {
ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
break;
}
// waiting for message
if (ttl >= 0) {
getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
getEntry(threadId).getLatch().acquire();
}
}
} finally {
unsubscribe(future, threadId);
}
// get(lockAsync(leaseTime, unit));
}
```
重點看下tryAcquire方法，把執行緒id作為一個引數傳遞進來，在這個方法裡面，找到tryLockInnerAsync方法點進去，
```
RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);",
Collections.