前言
我們知道Redis 緩存可以使用setNx來作為分布式鎖，但是我們直接使用setNx 需要考慮鎖過期的問題；此時我們可以使用Redisson 的lock 來實(shí)現(xiàn)分布式鎖，那么lock 內(nèi)部幫我們做了哪些工作呢。
一、Redisson 分布式鎖的實(shí)現(xiàn)：
1.1 引入redis 和 redisson jar
<!-- redis jar-->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.9.1</version>
</dependency>
1.2 redis 客戶端配置：
RedisConfig.java
package com.example.springredisms.config;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisPassword;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
@Configuration
@EnableCaching
public class RedisConfig {
@Value("${spring.redis.host:localhost}")
private String host;
@Value("${spring.redis.port:6379}")
private int port;
@Value("${spring.redis.database:0}")
private int db;
@Value("${spring.redis.password:null}")
private String password;
/**
* 配置lettuce連接池
*
* @return
*/
@ConfigurationProperties(prefix = "spring.redis.lettuce.pool")
public GenericObjectPoolConfig redisPool() {
return new GenericObjectPoolConfig<>();
}
/**
* 配置第二個數(shù)據(jù)源
*
* @return
*/
public RedisStandaloneConfiguration redisConfig() {
RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration(host, port);
redisStandaloneConfiguration.setDatabase(db);
if (password != null && !"".equals(password) && !"null".equals(password)){
redisStandaloneConfiguration.setPassword(RedisPassword.of(password));
}
return redisStandaloneConfiguration;
}
public LettuceConnectionFactory factory(GenericObjectPoolConfig redisPool, RedisStandaloneConfiguration redisConfig) {
LettuceClientConfiguration clientConfiguration = LettucePoolingClientConfiguration.builder().poolConfig(redisPool).build();
LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(redisConfig, clientConfiguration);
connectionFactory.afterPropertiesSet();
return connectionFactory;
}
/**
* 配置第一個數(shù)據(jù)源的RedisTemplate
* 注意：這里指定使用名稱=factory 的 RedisConnectionFactory
*
* @param
* @return
*/
@Bean("redisTemplate")
public RedisTemplate<String, Object> redisTemplate() {
RedisConnectionFactory factory1 = factory(redisPool(),redisConfig());
return redisTemplate(factory1);
}
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
// 使用Jackson2JsonRedisSerializer來序列化和反序列化redis的value值（默認(rèn)使用JDK的序列化方式）
Jackson2JsonRedisSerializer<Object> serializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
ObjectMapper objectMapper = new ObjectMapper();
// 指定要序列化的域，field,get和set,以及修飾符范圍，ANY是都有包括private和public
objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
// 指定序列化輸入的類型，類必須是非final修飾的，final修飾的類，比如String,Integer等會跑出異常
// objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
serializer.setObjectMapper(objectMapper);
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
// 配置連接工廠
redisTemplate.setConnectionFactory(redisConnectionFactory);
//使用StringRedisSerializer來序列化和反序列化redis的key值
redisTemplate.setKeySerializer(new StringRedisSerializer());
// 值采用json序列化
redisTemplate.setValueSerializer(serializer);
// 設(shè)置hash key 和value序列化模式
redisTemplate.setHashKeySerializer(new StringRedisSerializer());
redisTemplate.setHashValueSerializer(serializer);
redisTemplate.afterPropertiesSet();
return redisTemplate;
}
/***
* stringRedisTemplate默認(rèn)采用的是String的序列化策略
* @param redisConnectionFactory
* @return
*/
@Bean
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory){
StringRedisTemplate stringRedisTemplate = new StringRedisTemplate();
stringRedisTemplate.setConnectionFactory(redisConnectionFactory);
return stringRedisTemplate;
}
}
1.3 業(yè)務(wù)實(shí)現(xiàn)：
@Autowired
private RedissonClient redisson;
@Autowired
private RedisTemplate redisTemplate;
public String lockTest() {
String lockKey ="good:123";
// 初始化 lock
RLock lock = redisson.getLock(lockKey);
// 阻塞獲取鎖
lock.lock();
// 獲取鎖成功執(zhí)行扣減庫存邏輯
try{
Thread.sleep(5000);
String stockKey ="good:stock:123";
int stock = (Integer) redisTemplate.opsForValue().get(stockKey);
if (stock >0){
stock-=1;
redisTemplate.opsForValue().set(stockKey,stock);
}else{
return "商品已經(jīng)賣完了";
}
}catch (Exception ex){
}
finally {
// 釋放鎖
lock.unlock();
}
return "success";
}
二、Redisson lock 實(shí)現(xiàn)原理
2.1 lock.lock()：
lock.lock() 阻塞獲取 redis 鎖，獲取到鎖之后繼續(xù)向下執(zhí)行業(yè)務(wù)邏輯；
public void lock() {
try {
// 相應(yīng)中斷的獲取鎖
this.lockInterruptibly();
} catch (InterruptedException var2) {
Thread.currentThread().interrupt();
}
}
lockInterruptibly()：
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
long threadId = Thread.currentThread().getId();
// tryAcquire 獲取鎖，如果獲取成功 ttl 為null ，如果獲取失敗，說明當(dāng)前有線程在持有鎖，返回的是鎖的剩余時間
Long ttl = this.tryAcquire(leaseTime, unit, threadId);
if (ttl != null) {
// 當(dāng)前線程沒有獲取到鎖，訂閱一個與鎖相關(guān)聯(lián)的 Redis 頻道
// 當(dāng)鎖被釋放時，持有鎖的線程會向相關(guān)聯(lián)的 Redis 頻道發(fā)布一條消息。
// 那些訂閱了這個頻道并正在等待鎖釋放的線程會接收到這個消息，
// 并會再次嘗試獲取鎖。這個機(jī)制使得等待鎖的線程能立刻知道鎖何時被釋放。
RFuture<RedissonLockEntry> future = this.subscribe(threadId);
this.commandExecutor.syncSubscription(future);
try {
while(true) {
// 循環(huán)去獲取鎖
ttl = this.tryAcquire(leaseTime, unit, threadId);
if (ttl == null) {
// 獲取到鎖直接返回
return;
}
// 如果鎖的剩余時間還有很多
if (ttl >= 0L) {
// 嘗試去獲取鎖，如果失敗則加入到等待的雙向鏈表節(jié)點(diǎn)中，同時park 掛起當(dāng)前線程
this.getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
// 如果鎖已經(jīng)沒有剩余時間了，鎖到期了，則取獲取鎖
this.getEntry(threadId).getLatch().acquire();
}
}
} finally {
// 獲取鎖成功后 當(dāng)前線程取消訂閱這把鎖
this.unsubscribe(future, threadId);
}
}
}
tryAcquire() 嘗試去獲取鎖 ：
private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
return (Long)this.get(this.tryAcquireAsync(leaseTime, unit, threadId));
}
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
if (leaseTime != -1L) {
// 如果 不需要進(jìn)行鎖續(xù)命則 直接嘗試去獲取鎖
return this.tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
} else {
// 嘗試去獲取鎖
RFuture<Long> ttlRemainingFuture = this.tryLockInnerAsync(this.commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.addListener(new FutureListener<Long>() {
public void operationComplete(Future<Long> future) throws Exception {
if (future.isSuccess()) {
Long ttlRemaining = (Long)future.getNow();
if (ttlRemaining == null) {
// 獲取鎖成功，開啟一個定時任務(wù) 默認(rèn)每隔10 s 完成一次鎖續(xù)命
RedissonLock.this.scheduleExpirationRenewal(threadId);
}
}
}
});
return ttlRemainingFuture;
}
}
tryLockInnerAsync 獲取鎖：
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
this.internalLockLeaseTime = unit.toMillis(leaseTime);
return this.commandExecutor.evalWriteAsync(this.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.singletonList(this.getName()), new Object[]{this.internalLockLeaseTime, this.getLockName(threadId)});
}
這里完成了獲取鎖和重入鎖的實(shí)現(xiàn)：
- 如果當(dāng)前鎖沒有被線程占有，則設(shè)置當(dāng)前當(dāng)前線程 占有這把鎖，并設(shè)置鎖的過期時間為30s，返回null；
- 如果當(dāng)前 鎖已經(jīng)存在，并且是相同線程占用，則 設(shè)置將鎖的重入次數(shù)+1，并設(shè)置鎖的過期時間為30ss，返回null；
- 如果鎖已經(jīng)被其它線程占有 ，則返回鎖的過期時間；
scheduleExpirationRenewal ：開啟鎖續(xù)命的定時任務(wù)
private void scheduleExpirationRenewal(final long threadId) {
if (!expirationRenewalMap.containsKey(this.getEntryName())) {
// 延遲 internalLockLeaseTime / 3L 默認(rèn)值是 30/3 =10s 后開啟任務(wù)
Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
public void run(Timeout timeout) throws Exception {
// redis 鎖續(xù)期
RFuture<Boolean> future = RedissonLock.this.renewExpirationAsync(threadId);
future.addListener(new FutureListener<Boolean>() {
public void operationComplete(Future<Boolean> future) throws Exception {
RedissonLock.expirationRenewalMap.remove(RedissonLock.this.getEntryName());
if (!future.isSuccess()) {
// 當(dāng)前線程沒有獲取獲取到鎖，進(jìn)行延期則直接報錯
RedissonLock.log.error("Can't update lock " + RedissonLock.this.getName() + " expiration", future.cause());
} else {
// 續(xù)期成功后，繼續(xù)進(jìn)行下一次調(diào)用
if ((Boolean)future.getNow()) {
RedissonLock.this.scheduleExpirationRenewal(threadId);
}
}
}
});
}
}, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);
if (expirationRenewalMap.putIfAbsent(this.getEntryName(), new ExpirationEntry(threadId, task)) != null) {
task.cancel();
}
}
}
renewExpirationAsync 鎖續(xù)期：
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
return this.commandExecutor.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1)
then redis.call('pexpire', KEYS[1], ARGV[1]);
return 1; end; return 0;",
Collections.singletonList(this.getName()), new Object[]{this.internalLockLeaseTime, this.getLockName(threadId)});
}
- 如果這把鎖還存在，則設(shè)置過期時間（默認(rèn)值是30s）并且返回true
- 這把鎖不存在返回false；
2.2 鎖釋放 lock.unlock()：
在業(yè)務(wù)執(zhí)行完畢之后 通過lock.unlock(); 釋放鎖
public void unlock() {
try {
// 釋放 鎖
this.get(this.unlockAsync(Thread.currentThread().getId()));
} catch (RedisException var2) {
if (var2.getCause() instanceof IllegalMonitorStateException) {
throw (IllegalMonitorStateException)var2.getCause();
} else {
throw var2;
}
}
}
unlockAsync 釋放鎖：
public RFuture<Void> unlockAsync(final long threadId) {
final RPromise<Void> result = new RedissonPromise();
// 釋放鎖
RFuture<Boolean> future = this.unlockInnerAsync(threadId);
future.addListener(new FutureListener<Boolean>() {
public void operationComplete(Future<Boolean> future) throws Exception {
if (!future.isSuccess()) {
RedissonLock.this.cancelExpirationRenewal(threadId);
result.tryFailure(future.cause());
} else {
//
Boolean opStatus = (Boolean)future.getNow();
if (opStatus == null) {
IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + RedissonLock.this.id + " thread-id: " + threadId);
result.tryFailure(cause);
} else {
if (opStatus) {
// 釋放鎖成功 則取消鎖延期
RedissonLock.this.cancelExpirationRenewal((Long)null);
}
result.trySuccess((Object)null);
}
}
}
});
return result;
}
unlockInnerAsync 釋放鎖：文章來源：http://www.zghlxwxcb.cn/news/detail-824435.html
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return this.commandExecutor.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('exists', KEYS[1]) == 0)
then redis.call('publish', KEYS[2], ARGV[1]); return 1; end;
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then return nil;end;
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
if (counter > 0) then redis.call('pexpire', KEYS[1], ARGV[2]); return 0;
else redis.call('del', KEYS[1]);
redis.call('publish', KEYS[2], ARGV[1]); return 1; end;
return nil;", Arrays.asList(this.getName(), this.getChannelName()), new Object[]{LockPubSub.unlockMessage, this.internalLockLeaseTime, this.getLockName(threadId)});
}
- 如果當(dāng)前鎖已經(jīng)不存在，則發(fā)布鎖釋放消息，讓其他阻塞的現(xiàn)車去搶占鎖，返回true；
- 如果鎖還存在，則將鎖的重入次數(shù)-1 后返回鎖的重入次數(shù)；
- 如果重入次數(shù)大于0 說明當(dāng)前線程還需要績效持有鎖，重新設(shè)置鎖的過期時間，返回false；
- 如果重入次數(shù)等于0 ，則刪除這個key 相當(dāng)于是否了鎖，然后發(fā)布鎖釋放消息，讓其他阻塞的現(xiàn)車去搶占鎖，返回true；
總結(jié)
鎖過期自動續(xù)時的觸發(fā)條件是tryLock設(shè)置的鎖到期時間leaseTime == -1；自動續(xù)時原理就是創(chuàng)建一個定時任務(wù)，每internalLockLeaseTime / 3時間觸發(fā)一次，如果發(fā)現(xiàn)持有鎖未釋放，就把鎖過期時間更新為internalLockLeaseTime（internalLockLeaseTime的值取得是lockWatchdogTimeout默認(rèn)是30s）；過期時間更新成功后，再次遞歸調(diào)用renewExpiration()，創(chuàng)建下一次定時任務(wù)；默認(rèn)每次都延期30s；
這種機(jī)制主要是為了避免在沒來得及解鎖的情況下系統(tǒng)就掛了，導(dǎo)致該鎖在redis中一直都被占用，其他線程永遠(yuǎn)都無法獲取鎖，也就是死鎖的情況；文章來源地址http://www.zghlxwxcb.cn/news/detail-824435.html
到了這里，關(guān)于源碼篇--Redisson 分布式鎖lock的實(shí)現(xiàn)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
