Dalam persekitaran berbilang benang, untuk memastikan bahawa blok kod hanya boleh diakses oleh satu utas pada masa yang sama, di Java secara amnya kita boleh menggunakan sintaks yang disegerakkan dan ReetrantLock untuk memastikan bahawa ini sebenarnya Di atas adalah kaedah kunci tempatan. Tetapi kini syarikat menggunakan seni bina teragih Dalam persekitaran yang diedarkan, bagaimana untuk memastikan bahawa benang pada nod yang berbeza dilaksanakan secara serentak? Oleh itu, kunci yang diedarkan diperkenalkan, yang merupakan cara untuk mengawal akses yang saling eksklusif kepada sumber yang dikongsi antara sistem yang diedarkan. Dalam sistem yang diedarkan, berbilang perkhidmatan digunakan pada berbilang mesin Apabila pengguna pada klien memulakan permintaan pemasukan data, jika tiada jaminan mekanisme kunci yang diedarkan, maka perkhidmatan Berbilang boleh melakukan operasi pemasukan serentak. mengakibatkan pemasukan data berulang, yang boleh menyebabkan masalah bagi sesetengah perniagaan yang tidak membenarkan data berlebihan. Mekanisme kunci yang diedarkan adalah untuk menyelesaikan masalah seperti ini dan memastikan akses yang saling eksklusif kepada sumber yang dikongsi antara berbilang perkhidmatan Jika satu perkhidmatan merampas kunci yang diedarkan dan perkhidmatan lain tidak memperoleh kunci, tiada operasi seterusnya akan dilakukan. Maksud umum adalah seperti yang ditunjukkan dalam rajah di bawah:
Ciri-ciri kunci teragih
. Perintah pelaksanaan teras kaedah terakhir adalah seperti berikut: setnx+lua
- 获取锁（unique_value可以是UUID等）
SET resource_name unique_value NX PX 30000
- 释放锁（lua脚本中，一定要比较value，防止误解锁）
if redis.call("get",KEYS[1]) == ARGV[1] then
return redis.call("del",KEYS[1])
else
return 0
endset key value px milliseconds nx Kaedah pelaksanaan ini mempunyai tiga perkara utama (yang juga di mana kebarangkalian temu duga adalah sangat tinggi): Kunci diperoleh pada nod induk Redis; berlaku, dan nod hamba dinaik taraf kepada nod induk;Gunakan
- untuk perintah yang ditetapkan;
nilai mestilah unik; kunci, dan kunci tidak boleh disalah ertikan;
set key value px milliseconds nx- Malah, kelemahan terbesar kunci jenis ini ialah ia hanya bertindak pada satu nod Redis walaupun jika Redis memastikan ketersediaan tinggi melalui sentinel, jika nod induk disebabkan oleh beberapa. Jika suis induk-hamba berlaku atas sebab tertentu, kunci akan hilang:
Dapatkan masa Unix semasa dalam milisaat.
- Untuk mengelakkan satu titik masalah kegagalan, Redis author antirez mencadangkan kaedah pelaksanaan kunci teragih yang lebih maju berdasarkan persekitaran yang diedarkan:
Redlock
. Redlock juga merupakan satu-satunya cara di antara semua pelaksanaan kunci yang diedarkan dalam Redis yang boleh menjadikan penemuduga klimaks.- Kunci teragih lanjutan Redis: RedlockAlgoritma kunci merah yang dicadangkan oleh antirez mungkin seperti ini:
- Dalam persekitaran pengedaran Redis, kami mengandaikan bahawa terdapat N Redis master . Nod ini
adalah bebas sepenuhnya antara satu sama lain, dan tiada replikasi tuan-hamba atau mekanisme koordinasi kelompok lain
. Kami memastikan bahawa kami akan memperoleh dan melepaskan kunci pada N kejadian menggunakan kaedah yang sama seperti di bawah contoh tunggal Redis. Sekarang kita mengandaikan bahawa terdapat 5 nod induk Redis, dan kita perlu menjalankan kejadian Redis ini pada 5 pelayan untuk memastikan bahawa mereka tidak semua akan turun pada masa yang sama.- Untuk mendapatkan kunci, pelanggan hendaklah melakukan operasi berikut:
cuba memperoleh kunci daripada 5 kejadian dalam turutan, menggunakan kekunci yang sama dan
nilai unikialah 10 saat, tamat masa hendaklah antara 5-50 milisaat. Ini boleh mengelakkan situasi di mana Redis bahagian pelayan telah menutup telefon dan pelanggan masih menunggu keputusan respons. Jika pelayan tidak bertindak balas dalam masa yang ditetapkan, pelanggan harus cuba mendapatkan kunci daripada contoh Redis yang lain secepat mungkin.
客户端使用当前时间减去开始获取锁时间（步骤1记录的时间）就得到获取锁使用的时间。当且仅当从大多数（N/2+1，这里是3个节点）的Redis节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。
如果取到了锁，key的真正有效时间等于有效时间减去获取锁所使用的时间（步骤3计算的结果）。
如果因为某些原因，获取锁失败（没有在至少N/2+1个Redis实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的Redis实例上进行解锁（即便某些Redis实例根本就没有加锁成功，防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁）。
此处不讨论时钟漂移
redisson已经有对redlock算法封装，接下来对其用法进行简单介绍，并对核心源码进行分析（假设5个redis实例）。
<!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.3.2</version>
</dependency>首先，我们来看一下redission封装的redlock算法实现的分布式锁用法，非常简单，跟重入锁（ReentrantLock）有点类似：
Config config = new Config();
config.useSentinelServers().addSentinelAddress("127.0.0.1:6369","127.0.0.1:6379", "127.0.0.1:6389")
.setMasterName("masterName")
.setPassword("password").setDatabase(0);
RedissonClient redissonClient = Redisson.create(config);
// 还可以getFairLock(), getReadWriteLock()
RLock redLock = redissonClient.getLock("REDLOCK_KEY");
boolean isLock;
try {
isLock = redLock.tryLock();
// 500ms拿不到锁, 就认为获取锁失败。10000ms即10s是锁失效时间。
isLock = redLock.tryLock(500, 10000, TimeUnit.MILLISECONDS);
if (isLock) {
//TODO if get lock success, do something;
}
} catch (Exception e) {
} finally {
// 无论如何, 最后都要解锁
redLock.unlock();
}实现分布式锁的一个非常重要的点就是set的value要具有唯一性，redisson的value是怎样保证value的唯一性呢？答案是UUID+threadId。入口在redissonClient.getLock(“REDLOCK_KEY”)，源码在Redisson.java和RedissonLock.java中：
protected final UUID id = UUID.randomUUID();
String getLockName(long threadId) {
return id + ":" + threadId;
}获取锁的代码为redLock.tryLock()或者redLock.tryLock(500, 10000, TimeUnit.MILLISECONDS)，两者的最终核心源码都是下面这段代码，只不过前者获取锁的默认租约时间（leaseTime）是LOCK_EXPIRATION_INTERVAL_SECONDS，即30s：
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
// 获取锁时向5个redis实例发送的命令
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
// 首先分布式锁的KEY不能存在，如果确实不存在，那么执行hset命令（hset REDLOCK_KEY uuid+threadId 1），并通过pexpire设置失效时间（也是锁的租约时间）
"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; " +
// 如果分布式锁的KEY已经存在，并且value也匹配，表示是当前线程持有的锁，那么重入次数加1，并且设置失效时间
"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; " +
// 获取分布式锁的KEY的失效时间毫秒数
"return redis.call('pttl', KEYS[1]);",
// 这三个参数分别对应KEYS[1]，ARGV[1]和ARGV[2]
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}获取锁的命令中，
KEYS[1]就是Collections.singletonList(getName())，表示分布式锁的key，即REDLOCK_KEY；
ARGV[1]就是internalLockLeaseTime，即锁的租约时间，默认30s；
ARGV[2]就是getLockName(threadId)，是获取锁时set的唯一值，即UUID+threadId：
释放锁的代码为redLock.unlock()，核心源码如下：
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
// 向5个redis实例都执行如下命令
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
// 如果分布式锁KEY不存在，那么向channel发布一条消息
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end;" +
// 如果分布式锁存在，但是value不匹配，表示锁已经被占用，那么直接返回
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
// 如果就是当前线程占有分布式锁，那么将重入次数减1
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
// 重入次数减1后的值如果大于0，表示分布式锁有重入过，那么只设置失效时间，还不能删除
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
"else " +
// 重入次数减1后的值如果为0，表示分布式锁只获取过1次，那么删除这个KEY，并发布解锁消息
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; "+
"end; " +
"return nil;",
// 这5个参数分别对应KEYS[1]，KEYS[2]，ARGV[1]，ARGV[2]和ARGV[3]
Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));
}下面利用SpringBoot + Jedis + AOP的组合来实现一个简易的分布式锁。
自定义一个注解，被注解的方法会执行获取分布式锁的逻辑
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface RedisLock {
/**
* 业务键
*
* @return
*/
String key();
/**
* 锁的过期秒数,默认是5秒
*
* @return
*/
int expire() default 5;
/**
* 尝试加锁，最多等待时间
*
* @return
*/
long waitTime() default Long.MIN_VALUE;
/**
* 锁的超时时间单位
*
* @return
*/
TimeUnit timeUnit() default TimeUnit.SECONDS;
}在AOP中我们去执行获取分布式锁和释放分布式锁的逻辑，代码如下：
@Aspect
@Component
public class LockMethodAspect {
@Autowired
private RedisLockHelper redisLockHelper;
@Autowired
private JedisUtil jedisUtil;
private Logger logger = LoggerFactory.getLogger(LockMethodAspect.class);
@Around("@annotation(com.redis.lock.annotation.RedisLock)")
public Object around(ProceedingJoinPoint joinPoint) {
Jedis jedis = jedisUtil.getJedis();
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
Method method = signature.getMethod();
RedisLock redisLock = method.getAnnotation(RedisLock.class);
String value = UUID.randomUUID().toString();
String key = redisLock.key();
try {
final boolean islock = redisLockHelper.lock(jedis,key, value, redisLock.expire(), redisLock.timeUnit());
logger.info("isLock : {}",islock);
if (!islock) {
logger.error("获取锁失败");
throw new RuntimeException("获取锁失败");
}
try {
return joinPoint.proceed();
} catch (Throwable throwable) {
throw new RuntimeException("系统异常");
}
} finally {
logger.info("释放锁");
redisLockHelper.unlock(jedis,key, value);
jedis.close();
}
}
}@Component
public class RedisLockHelper {
private long sleepTime = 100;
/**
* 直接使用setnx + expire方式获取分布式锁
* 非原子性
*
* @param key
* @param value
* @param timeout
* @return
*/
public boolean lock_setnx(Jedis jedis,String key, String value, int timeout) {
Long result = jedis.setnx(key, value);
// result = 1时，设置成功，否则设置失败
if (result == 1L) {
return jedis.expire(key, timeout) == 1L;
} else {
return false;
}
}
/**
* 使用Lua脚本，脚本中使用setnex+expire命令进行加锁操作
*
* @param jedis
* @param key
* @param UniqueId
* @param seconds
* @return
*/
public boolean Lock_with_lua(Jedis jedis,String key, String UniqueId, int seconds) {
String lua_scripts = "if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then" +
"redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end";
List<String> keys = new ArrayList<>();
List<String> values = new ArrayList<>();
keys.add(key);
values.add(UniqueId);
values.add(String.valueOf(seconds));
Object result = jedis.eval(lua_scripts, keys, values);
//判断是否成功
return result.equals(1L);
}
/**
* 在Redis的2.6.12及以后中,使用 set key value [NX] [EX] 命令
*
* @param key
* @param value
* @param timeout
* @return
*/
public boolean lock(Jedis jedis,String key, String value, int timeout, TimeUnit timeUnit) {
long seconds = timeUnit.toSeconds(timeout);
return "OK".equals(jedis.set(key, value, "NX", "EX", seconds));
}
/**
* 自定义获取锁的超时时间
*
* @param jedis
* @param key
* @param value
* @param timeout
* @param waitTime
* @param timeUnit
* @return
* @throws InterruptedException
*/
public boolean lock_with_waitTime(Jedis jedis,String key, String value, int timeout, long waitTime,TimeUnit timeUnit) throws InterruptedException {
long seconds = timeUnit.toSeconds(timeout);
while (waitTime >= 0) {
String result = jedis.set(key, value, "nx", "ex", seconds);
if ("OK".equals(result)) {
return true;
}
waitTime -= sleepTime;
Thread.sleep(sleepTime);
}
return false;
}
/**
* 错误的解锁方法—直接删除key
*
* @param key
*/
public void unlock_with_del(Jedis jedis,String key) {
jedis.del(key);
}
/**
* 使用Lua脚本进行解锁操纵，解锁的时候验证value值
*
* @param jedis
* @param key
* @param value
* @return
*/
public boolean unlock(Jedis jedis,String key,String value) {
String luaScript = "if redis.call('get',KEYS[1]) == ARGV[1] then " +
"return redis.call('del',KEYS[1]) else return 0 end";
return jedis.eval(luaScript, Collections.singletonList(key), Collections.singletonList(value)).equals(1L);
}
}定义一个TestController来测试我们实现的分布式锁
@RestController
public class TestController {
@RedisLock(key = "redis_lock")
@GetMapping("/index")
public String index() {
return "index";
}
}Atas ialah kandungan terperinci Analisis contoh kunci yang diedarkan Redis. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!
Perisian pangkalan data yang biasa digunakan
Apakah pangkalan data dalam memori?
Yang manakah mempunyai kelajuan bacaan yang lebih pantas, mongodb atau redis?
Cara menggunakan redis sebagai pelayan cache
Bagaimana redis menyelesaikan ketekalan data
Bagaimanakah mysql dan redis memastikan konsistensi penulisan dua kali?
Apakah data yang biasanya disimpan oleh redis cache?
Apakah 8 jenis data redis
![[Web front-end] Permulaan pantas Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
