首頁 > 軟體

Redis分散式鎖的實現方式

2023-04-04 06:01:50

一、分散式鎖是什麼

分散式鎖是 滿足分散式系統或叢集模式下多程序可見並且互斥的鎖。

基於Redis實現分散式鎖：

1、獲取鎖

  • 互斥：確保只能有一個執行緒獲取鎖;
  • 非阻塞：嘗試獲取鎖，成功返回true，失敗返回false；

新增鎖過期時間，避免服務宕機引起死鎖。

SET lock thread1 NX EX 10

2、釋放鎖

  • 手動釋放；DEL key1
  • 超時釋放，獲取鎖時新增一個超時鎖；

二、程式碼範例

package com.guor.utils;

import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.concurrent.TimeUnit;

public class RedisLock implements ILock{

    private String name;
    private StringRedisTemplate stringRedisTemplate;

    public RedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    private static final String KEY_PREFIX = "lock:";

    @Override
    public boolean tryLock(long timeout) {
        // 獲取執行緒唯一標識
        long threadId = Thread.currentThread().getId();
        // 獲取鎖
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId+"", timeout, TimeUnit.SECONDS);
        // 防止拆箱的空指標異常
        return Boolean.TRUE.equals(success);
    }

    @Override
    public void unlock() {
        stringRedisTemplate.delete(KEY_PREFIX + name);
    }
}

上面程式碼存在鎖誤刪問題：

  1. 如果執行緒1獲取鎖，但執行緒1發生了阻塞，導致Redis超時釋放鎖；
  2. 此時，執行緒2嘗試獲取鎖，成功，並執行業務；
  3. 此時，執行緒1重新開始執行任務，並執行完畢，執行釋放鎖（即刪除鎖）；
  4. 但是，執行緒1刪除的鎖，和執行緒2的鎖是同一把鎖，這就是分散式鎖誤刪問題；

在釋放鎖時，釋放執行緒自己的分散式鎖，就可以解決這個問題。

package com.guor.utils;

import cn.hutool.core.lang.UUID;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.concurrent.TimeUnit;

public class RedisLock implements ILock{

    private String name;
    private StringRedisTemplate stringRedisTemplate;

    public RedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    private static final String KEY_PREFIX = "lock:";
    private static final String UUID_PREFIX = UUID.randomUUID().toString(true) + "-";

    @Override
    public boolean tryLock(long timeout) {
        // 獲取執行緒唯一標識
        String threadId = UUID_PREFIX + Thread.currentThread().getId();
        // 獲取鎖
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId, timeout, TimeUnit.SECONDS);
        // 防止拆箱的空指標異常
        return Boolean.TRUE.equals(success);
    }

    @Override
    public void unlock() {
        // 獲取執行緒唯一標識
        String threadId = UUID_PREFIX + Thread.currentThread().getId();
        // 獲取鎖中的標識
        String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
        // 判斷標示是否一致
        if(threadId.equals(id)) {
            // 釋放鎖
            stringRedisTemplate.delete(KEY_PREFIX + name);
        }
    }
}

三、基於SETNX實現的分散式鎖存在下面幾個問題

1、不可重入

同一個執行緒無法多次獲取同一把鎖。

2、不可重試

獲取鎖只嘗試一次就返回false，沒有重試機制。

3、超時釋放

鎖的超時釋放雖然可以避免死鎖，但如果業務執行耗時較長，也會導致鎖釋放，存在安全隱患。

4、主從一致性

如果Redis是叢集部署的，主從同步存在延遲，當主機宕機時，此時會選一個從作為主機，但是此時的從沒有鎖標識，此時，其它執行緒可能會獲取到鎖，導致安全問題。

四、Redisson實現分散式鎖

Redisson是一個在Redis的基礎上實現的Java駐記憶體資料網格。它不僅提供了一系列的分散式的Java常用物件，還提供了許多分散式服務，其中包含各種分散式鎖的實現。

1、pom

<!--redisson-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.6</version>
</dependency>

2、設定類

package com.guor.config;

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient(){
        // 設定
        Config config = new Config();

        /**
         * 單點地址useSingleServer，叢集地址useClusterServers
         */
        config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        // 建立RedissonClient物件
        return Redisson.create(config);
    }
}

3、測試類

package com.guor;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.boot.test.context.SpringBootTest;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

@Slf4j
@SpringBootTest
class RedissonTest {

    @Resource
    private RedissonClient redissonClient;

    private RLock lock;

    @BeforeEach
    void setUp() {
    	// 獲取指定名稱的鎖
        lock = redissonClient.getLock("nezha");
    }

    @Test
    void test() throws InterruptedException {
        // 嘗試獲取鎖
        boolean isLock = lock.tryLock(1L, TimeUnit.SECONDS);
        if (!isLock) {
            log.error("獲取鎖失敗");
            return;
        }
        try {
            log.info("哪吒最帥，哈哈哈");
        } finally {
            // 釋放鎖
            lock.unlock();
        }
    }
}

五、探索tryLock原始碼

1、tryLock原始碼

嘗試獲取鎖

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
	// 最大等待時間
	long time = unit.toMillis(waitTime);
	long current = System.currentTimeMillis();
	long threadId = Thread.currentThread().getId();
	Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
	if (ttl == null) {
		return true;
	} else {
		// 剩餘等待時間 = 最大等待時間 - 獲取鎖失敗消耗的時間
		time -= System.currentTimeMillis() - current;
		if (time <= 0L) {// 獲取鎖失敗
			this.acquireFailed(waitTime, unit, threadId);
			return false;
		} else {
			// 再次嘗試獲取鎖
			current = System.currentTimeMillis();
			// subscribe訂閱其它釋放鎖的訊號
			RFuture<RedissonLockEntry> subscribeFuture = this.subscribe(threadId);
			// 當Future在等待指定時間time內完成時，返回true
			if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
				if (!subscribeFuture.cancel(false)) {
					subscribeFuture.onComplete((res, e) -> {
						if (e == null) {
							// 取消訂閱
							this.unsubscribe(subscribeFuture, threadId);
						}

					});
				}

				this.acquireFailed(waitTime, unit, threadId);
				return false;// 獲取鎖失敗
			} else {
				try {
					// 剩餘等待時間 = 剩餘等待時間 - 獲取鎖失敗消耗的時間
					time -= System.currentTimeMillis() - current;
					if (time <= 0L) {
						this.acquireFailed(waitTime, unit, threadId);
						boolean var20 = false;
						return var20;
					} else {
						boolean var16;
						do {
							long currentTime = System.currentTimeMillis();
							// 重試獲取鎖
							ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
							if (ttl == null) {
								var16 = true;
								return var16;
							}
							// 再次失敗了，再看一下剩餘時間
							time -= System.currentTimeMillis() - currentTime;
							if (time <= 0L) {
								this.acquireFailed(waitTime, unit, threadId);
								var16 = false;
								return var16;
							}
							// 再重試獲取鎖
							currentTime = System.currentTimeMillis();
							if (ttl >= 0L && ttl < time) {
								// 通過號誌的方式嘗試獲取訊號，如果等待時間內，依然沒有結果，會返回false
								((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
							} else {
								((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
							}
							time -= System.currentTimeMillis() - currentTime;
						} while(time > 0L);

						this.acquireFailed(waitTime, unit, threadId);
						var16 = false;
						return var16;
					}
				} finally {
					this.unsubscribe(subscribeFuture, threadId);
				}
			}
		}
	}
}

2、重置鎖的有效期

private void scheduleExpirationRenewal(long threadId) {
	RedissonLock.ExpirationEntry entry = new RedissonLock.ExpirationEntry();
	// this.getEntryName()：鎖的名字，一個鎖對應一個entry
	// putIfAbsent：如果不存在，將鎖和entry放到map裡
	RedissonLock.ExpirationEntry oldEntry = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);
	if (oldEntry != null) {
		// 同一個執行緒多次獲取鎖，相當於重入
		oldEntry.addThreadId(threadId);
	} else {
		// 如果是第一次
		entry.addThreadId(threadId);
		// 更新有效期
		this.renewExpiration();
	}
}

更新有效期，遞迴呼叫更新有效期，永不過期

private void renewExpiration() {
	// 從map中得到當前鎖的entry
	RedissonLock.ExpirationEntry ee = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
	if (ee != null) {
		// 開啟延時任務
		Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
			public void run(Timeout timeout) throws Exception {
				RedissonLock.ExpirationEntry ent = (RedissonLock.ExpirationEntry)RedissonLock.EXPIRATION_RENEWAL_MAP.get(RedissonLock.this.getEntryName());
				if (ent != null) {
					// 取出執行緒id
					Long threadId = ent.getFirstThreadId();
					if (threadId != null) {
						// 重新整理有效期
						RFuture<Boolean> future = RedissonLock.this.renewExpirationAsync(threadId);
						future.onComplete((res, e) -> {
							if (e != null) {
								RedissonLock.log.error("Can't update lock " + RedissonLock.this.getName() + " expiration", e);
							} else {
								if (res) {
									// 遞迴呼叫更新有效期，永不過期
									RedissonLock.this.renewExpiration();
								}
							}
						});
					}
				}
			}
		}, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);// 10S
		ee.setTimeout(task);
	}
}

更新有效期

protected RFuture<Boolean> renewExpirationAsync(long threadId) {
	return this.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()), this.internalLockLeaseTime, this.getLockName(threadId));
}

3、呼叫lua指令碼

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
	// 鎖釋放時間
	this.internalLockLeaseTime = unit.toMillis(leaseTime);
	return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, command, 
		// 判斷鎖成功
		"if (redis.call('exists', KEYS[1]) == 0) then
			redis.call('hincrby', KEYS[1], ARGV[2], 1); // 如果不存在，記錄鎖標識，次數+1
			redis.call('pexpire', KEYS[1], ARGV[1]); // 設定鎖有效期
			return nil; // 相當於Java的null
		end; 
		if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then 
			redis.call('hincrby', KEYS[1], ARGV[2], 1); // 如果存在，判斷鎖標識是否是自己的，次數+1
			redis.call('pexpire', KEYS[1], ARGV[1]); // 設定鎖有效期
			return nil; 
		end; 
		// 判斷鎖失敗，pttl：指定鎖剩餘有效期，單位毫秒，KEYS[1]：鎖的名稱
		return redis.call('pttl', KEYS[1]);", 
			Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId));
}

六、釋放鎖unlock原始碼

1、取消更新任務

public RFuture<Void> unlockAsync(long threadId) {
	RPromise<Void> result = new RedissonPromise();
	RFuture<Boolean> future = this.unlockInnerAsync(threadId);
	future.onComplete((opStatus, e) -> {
		// 取消更新任務
		this.cancelExpirationRenewal(threadId);
		if (e != null) {
			result.tryFailure(e);
		} else if (opStatus == null) {
			IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + this.id + " thread-id: " + threadId);
			result.tryFailure(cause);
		} else {
			result.trySuccess((Object)null);
		}
	});
	return result;
}

2、刪除定時任務

void cancelExpirationRenewal(Long threadId) {
	// 從map中取出當前鎖的定時任務entry
	RedissonLock.ExpirationEntry task = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
	if (task != null) {
		if (threadId != null) {
			task.removeThreadId(threadId);
		}
		// 刪除定時任務
		if (threadId == null || task.hasNoThreads()) {
			Timeout timeout = task.getTimeout();
			if (timeout != null) {
				timeout.cancel();
			}

			EXPIRATION_RENEWAL_MAP.remove(this.getEntryName());
		}
	}
}

以上就是Redis分散式鎖的實現方式的詳細內容，更多關於Redis實現分散式鎖的資料請關注it145.com其它相關文章！


相關文章

  • 德漢姆中學就讀條件 德漢姆中學就讀條件

    <em>Mac</em>Book项目 2009年学校开始实施<em>Mac</em>Book项目,所有师生配备一本<em>Mac</em>Book,并同步更新了校园无线网络。学校每周进行电脑技术更新,每月发送技术支持资料,极大改变了教学及学习方式。因此2011

    2021-06-01 09:32:01

  • 黑科技的輕簡出行，告別「電量焦慮症」：Anker氮化鎵超能充系列 黑科技的輕簡出行，告別「電量焦慮症」：Anker氮化鎵超能充系列

    综合看Anker超能充系列的性价比很高，并且与不仅和iPhone12/苹果<em>Mac</em>Book很配，而且适合多设备充电需求的日常使用或差旅场景，不管是安卓还是Switch同样也能用得上它，希望这次分享能给准备购入充电器的小伙伴们有所

    2021-06-01 09:31:42

  • 吳亦凡廠牌首秀，L4WUDU居然忘詞了，Rapper和明星比還是有差距！ 吳亦凡廠牌首秀，L4WUDU居然忘詞了，Rapper和明星比還是有差距！

    除了L4WUDU与吴亦凡已经多次共事，成为了明面上的厂牌成员，吴亦凡还曾带领20XXCLUB全队参加2020年的一场音乐节，这也是20XXCLUB首次全员合照，王嗣尧Turbo、陈彦希Regi、<em>Mac</em> Ova Seas、林渝植等人全部出场。然而让

    2021-06-01 09:31:34

  • IPFS、Chia、Bzz和ICP挖礦該怎麼選？哪一個更有優勢？ IPFS、Chia、Bzz和ICP挖礦該怎麼選？哪一個更有優勢？

    目前应用IPFS的机构：1 谷歌<em>浏览器</em>支持IPFS分布式协议 2 万维网 (历史档案博物馆)数据库 3 火狐<em>浏览器</em>支持 IPFS分布式协议 4 EOS 等数字货币数据存储 5 美国国会图书馆，历史资料永久保存在 IPFS 6 加

    2021-06-01 09:31:24

  • 有哪些事是買了雪佛蘭後才知道的？美事偷著樂，開拓者車主隨聊 有哪些事是買了雪佛蘭後才知道的？美事偷著樂，開拓者車主隨聊

    开拓者的车机是兼容苹果和<em>安卓</em>，虽然我不怎么用，但确实兼顾了我家人的很多需求：副驾的门板还配有解锁开关，有的时候老婆开车，下车的时候偶尔会忘记解锁，我在副驾驶可以自己开门：第二排设计很好，不仅配置了一个很大的

    2021-06-01 09:30:48

  • iPhone12在618最新定價，跌價幅度超過1400元，還等iPhone13嗎？ iPhone12在618最新定價，跌價幅度超過1400元，還等iPhone13嗎？

    不仅是<em>安卓</em>手机，苹果手机的降价力度也是前所未有了，iPhone12也“跳水价”了，发布价是6799元，如今已经跌至5308元，降价幅度超过1400元，最新定价确认了。iPhone12是苹果首款5G手机，同时也是全球首款5nm芯片的智能机，它

    2021-06-01 09:30:45

IT145.com E-mail:sddin#qq.com