Redis分散ロックを実装するにはどのような方法がありますか?

1. 分散ロックとは何ですか?

分散ロックは、分散システムまたはクラスター モードの複数のプロセスに表示されます。相互排他ロック。

Redis に基づく分散ロックの実装:

1. ロックの取得

• 相互排他: 1 つのスレッドのみがロックを取得できるようにします。

• ノンブロッキング: ロックの取得を試み、成功した場合は true、失敗した場合は false を返します。

ロックの有効期限を追加して、次のような原因によるデッドロックを回避します。サービスのダウンタイム。

SET ロック スレッド 1 NX EX 10

2. ロックの解除

• 手動リリース;DEL キー 1

• タイムアウト解除、ロック取得時にタイムアウトロックを追加;

2. コード例

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);
        }
    }
}

3.

SETNX に基づいて実装された分散ロックには次の問題があります

1. 再入不可

同じスレッドを複数回使用することはできません同じロックを取得します。

2. 再試行なし

ロックの取得を一度試行すると false が返されるため、再試行メカニズムはありません。

3. タイムアウト解除

タイムアウトでロックを解除するとデッドロックは回避できますが、業務実行に時間がかかる場合にはロックも解除されてしまい、セキュリティリスクが発生します。

4. マスター/スレーブの一貫性

Redis がクラスターにデプロイされている場合、マスターとスレーブの同期に遅延が発生し、ホストがダウンすると、スレーブがマスターとして選択されます。ただし、現時点ではロック識別子は存在しないため、他のスレッドがロックを取得する可能性があり、セキュリティ上の問題が発生します。

4. 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();
        }
    }
}

5. 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&#39;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(&#39;hexists&#39;, KEYS[1], ARGV[2]) == 1) then 
		// 更新有效期
		redis.call(&#39;pexpire&#39;, 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(&#39;exists&#39;, KEYS[1]) == 0) then
			redis.call(&#39;hincrby&#39;, KEYS[1], ARGV[2], 1); // 如果不存在，记录锁标识，次数+1
			redis.call(&#39;pexpire&#39;, KEYS[1], ARGV[1]); // 设置锁有效期
			return nil; // 相当于Java的null
		end; 
		if (redis.call(&#39;hexists&#39;, KEYS[1], ARGV[2]) == 1) then 
			redis.call(&#39;hincrby&#39;, KEYS[1], ARGV[2], 1); // 如果存在，判断锁标识是否是自己的，次数+1
			redis.call(&#39;pexpire&#39;, KEYS[1], ARGV[1]); // 设置锁有效期
			return nil; 
		end; 
		// 判断锁失败，pttl：指定锁剩余有效期，单位毫秒，KEYS[1]：锁的名称
		return redis.call(&#39;pttl&#39;, KEYS[1]);", 
			Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId));
}

6. ロック解除ソースコード

1. アップデートをキャンセルtask

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分散ロックを実装するにはどのような方法がありますか?の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。