Cara memperbaharui kunci yang diedarkan Redis
Postur kunci yang diedarkan Redis yang betul
Mengikut pemahaman Fei Chao, apabila ramai pelajar menggunakan kunci yang diedarkan, mereka terus mencari di Baidu untuk mencari satu Kelas alat kunci yang diedarkan Redis digunakan secara langsung. Kuncinya ialah kelas alat ini juga diisi dengan banyak System.out.println(); alat pelanggan redisson. Secara khusus Memperkenalkan github laman web temu janji gay terbesar yang boleh dicari.
Cara menjawab
Pertama sekali, jika anda telah menggunakan kunci yang diedarkan Redis dengan betul sebelum ini dan membaca rasmi yang berkaitan dokumen, soalan ini Sangat mudah. 🎜>
Secara lalai tamat masa pengawas kunci ialah 30 saat dan boleh ditukar melalui tetapan Config.lockWatchdogTimeout.
Tetapi jika anda melihat dokumen Cina
Tamat masa lalai untuk kunci semakan anjing pemerhati ialah 30 saat
Ayat Fei Chao ini ialah ayat samar-samar dari perspektif linguistik Ia mempunyai dua maksud
1 menyemak setiap 30 saat secara lalai Kunci masa tamat 2. Anjing pemerhati akan menyemak tamat masa kunci Masa kunci lalai ialah 30 saat
Melihat ini, saya harap semua orang tidak akan menggodam saya. . Guru pendidikan jasmani sekolah rendah, walaupun dia dan guru Cina adalah orang yang sama Jika bahasa Cina tidak bagus, kita boleh membuat kod sumber!
Analisis kod sumber
Kami menulis contoh terakhir berdasarkan contoh yang diberikan dalam dokumen rasmi Demo mudah, contoh adalah berdasarkan operasi gelombang Ctr+C dan Ctr+V dalam tangkapan skrin di atas, seperti berikut
public class DemoMain {
public static void main(String[] args) throws Exception {
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);
RLock lock = redisson.getLock("anyLock");
lock.lock();
//lock.unlock();
}
}Salin selepas log masuk
buat
Dari sini kita tahu bahawa dua parameter internalLockLeaseTime dan lockWatchdogTimeout adalah sama.Nilai lalai lockWatchdogTimeout adalah seperti berikut
public class Config {
private long lockWatchdogTimeout = 30 * 1000;
public long getLockWatchdogTimeout() {
return lockWatchdogTimeout;
}
//省略无关代码
}Salin selepas log masuk
Ia juga boleh dilihat daripada perkataan internalLockLeaseTime bahawa tamat masa kunci yang diedarkan ini Masa lalai ialah 30 saat Tetapi terdapat satu lagi soalan, iaitu, berapa kerap anjing pemantau ini memanjangkan tempoh sah? 🎜>
kunci
Kami tahu dari bingkai dalam gambar saya bahawa berjaya memperoleh kunci akan memulakan tugas yang dijadualkan, iaitu, pengawas . Tugasan berjadual akan menyemak secara berkala untuk memperbaharui renewExpirationAsync(threadId).
HashedWheelTimer dalam pakej netty-common digunakan di sini untuk masa Akaun rasmi Feichao telah mewujudkan hubungan kerjasama yang rapat dengan enjin carian utama kelas ini dalam mana-mana enjin carian untuk mengetahui parameter API yang berkaitan Maksudnya.Kami faham daripada gambar bahawa perbezaan masa antara setiap panggilan jadual yang dijadualkan ialah internalLockLeaseTime / 3. Iaitu 10 saat.
Kebenaran terbongkar
Melalui analisis kod sumber, kita tahu bahawa, Secara lalai, masa penguncian adalah 30 saat Jika perniagaan yang dikunci belum selesai, maka apabila 30-10 = 20 saat, pembaharuan. akan dilakukan dan kunci akan ditetapkan semula kepada 30 saat Pada masa ini Sesetengah pelajar mungkin bertanya lagi, bagaimana jika mesin perniagaan turun, tugas yang dijadualkan tidak dapat diperbaharui kunci akan dibuka selepas 30 saat
Semula 5 perangkap kunci yang diedarkan
1 kesilapan yang saya lakukan di atas Memandangkan utas semasa memperoleh kunci redis, memproses Kunci tidak dilepaskan dalam masa selepas perniagaan selesai, menyebabkan utas lain terus mencuba untuk mendapatkan kunci dan menyekat laporkan mesej ralat berikut
redis.clients.jedis.exceptions.JedisConnectionException: Tidak boleh mendapatkan sumber daripada kolam
Kolam benang redis tidak mempunyai utas percuma untuk memproses arahan klien.
Penyelesaian juga sangat mudah selagi kita berhati-hati, benang yang mendapat kunci melepaskan kunci dalam masa selepas memproses perniagaan Jika ia adalah kunci reentrant tetapi tidak mendapat kunci benang boleh melepaskan sambungan semasa dan tidur untuk tempoh masa.
public void lock() {
while (true) {
boolean flag = this.getLock(key);
if (flag) {
TODO .........
} else {
// 释放当前redis连接
redis.close();
// 休眠1000毫秒
sleep(1000);
}
}
}Salin selepas log masuk
2. Kunci B dilepaskan oleh A
Kami tahu bahawa prinsip pelaksanaan kunci Redis terletak pada arahan SETNX. Apabila kunci tidak wujud, nilai kunci ditetapkan kepada nilai dan nilai pulangan ialah 1 jika kunci yang diberikan sudah wujud, SETNX tidak mengambil sebarang tindakan dan nilai pulangan ialah 0; SETNX key value
Salin selepas log masuk
Jika logik perniagaan memakan masa pada masa ini dan masa pelaksanaan telah melebihi masa tamat tempoh kunci redis, maka kunci benang A dilepaskan secara automatik (kunci dipadamkan) dan benang B mengesan bahawa kunci myLock tidak wujud dan melaksanakan SETNX Perintah itu juga mendapat kunci.
Walau bagaimanapun, walaupun utas A telah melengkapkan logik perniagaan, kunci akan tetap dilepaskan (iaitu kunci akan dipadamkan), jadi kunci utas B juga akan dilepaskan oleh utas A.
Untuk mengelakkan situasi di atas, kami secara amnya perlu membawa nilai uniknya sendiri untuk mengenal pasti setiap utas semasa mengunci, dan hanya melepaskan kunci dengan nilai yang ditentukan, jika tidak, akan ada adegan huru-hara pelepasan kunci .
三、数据库事务超时
emm~ 聊redis锁咋还扯到数据库事务上来了？别着急往下看，看下边这段代码：
@Transaction
public void lock() {
while (true) {
boolean flag = this.getLock(key);
if (flag) {
insert();
}
}
}Salin selepas log masuk
给这个方法添加一个@Transaction注解开启事务，如代码中抛出异常进行回滚，要知道数据库事务可是有超时时间限制的，并不会无条件的一直等一个耗时的数据库操作。
比如：我们解析一个大文件，再将数据存入到数据库，如果执行时间太长，就会导致事务超时自动回滚。
一旦你的key长时间获取不到锁，获取锁等待的时间远超过数据库事务超时时间，程序就会报异常。
一般为解决这种问题，我们就需要将数据库事务改为手动提交、回滚事务。
@Autowired
DataSourceTransactionManager dataSourceTransactionManager;
@Transaction
public void lock() {
//手动开启事务
TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
try {
while (true) {
boolean flag = this.getLock(key);
if (flag) {
insert();
//手动提交事务
dataSourceTransactionManager.commit(transactionStatus);
}
}
} catch (Exception e) {
//手动回滚事务
dataSourceTransactionManager.rollback(transactionStatus);
}
}Salin selepas log masuk
四、锁过期了，业务还没执行完
这种情况和我们上边提到的第二种比较类似，但解决思路上略有不同。
同样是redis分布式锁过期，而业务逻辑没执行完的场景，不过，这里换一种思路想问题，把redis锁的过期时间再弄长点不就解决了吗？
那还是有问题，我们可以在加锁的时候，手动调长redis锁的过期时间，可这个时间多长合适？业务逻辑的执行时间是不可控的，调的过长又会影响操作性能。
要是redis锁的过期时间能够自动续期就好了。
为了解决这个问题我们使用redis客户端redisson，redisson很好的解决了redis在分布式环境下的一些棘手问题，它的宗旨就是让使用者减少对Redis的关注，将更多精力用在处理业务逻辑上。
redisson对分布式锁做了很好封装，只需调用API即可。
RLock lock = redissonClient.getLock("stockLock");Salin selepas log masuk
redisson在加锁成功后，会注册一个定时任务监听这个锁，每隔10秒就去查看这个锁，如果还持有锁，就对过期时间进行续期。默认过期时间30秒。这个机制也被叫做：“看门狗”，这名字。。。
举例子：假如加锁的时间是30秒，过10秒检查一次，一旦加锁的业务没有执行完，就会进行一次续期，把锁的过期时间再次重置成30秒。
通过分析下边redisson的源码实现可以发现，不管是加锁、解锁、续约都是客户端把一些复杂的业务逻辑，通过封装在Lua脚本中发送给redis，保证这段复杂业务逻辑执行的原子性。
@Slf4j
@Service
public class RedisDistributionLockPlus {
/**
* 加锁超时时间，单位毫秒， 即：加锁时间内执行完操作，如果未完成会有并发现象
*/
private static final long DEFAULT_LOCK_TIMEOUT = 30;
private static final long TIME_SECONDS_FIVE = 5 ;
/**
* 每个key的过期时间 {@link LockContent}
*/
private Map<String, LockContent> lockContentMap = new ConcurrentHashMap<>(512);
/**
* redis执行成功的返回
*/
private static final Long EXEC_SUCCESS = 1L;
/**
* 获取锁lua脚本， k1：获锁key, k2：续约耗时key, arg1:requestId，arg2：超时时间
*/
private static final String LOCK_SCRIPT = "if redis.call('exists', KEYS[2]) == 1 then ARGV[2] = math.floor(redis.call('get', KEYS[2]) + 10) end " +
"if redis.call('exists', KEYS[1]) == 0 then " +
"local t = redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) " +
"for k, v in pairs(t) do " +
"if v == 'OK' then return tonumber(ARGV[2]) end " +
"end " +
"return 0 end";
/**
* 释放锁lua脚本, k1：获锁key, k2：续约耗时key, arg1:requestId，arg2：业务耗时 arg3: 业务开始设置的timeout
*/
private static final String UNLOCK_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
"local ctime = tonumber(ARGV[2]) " +
"local biz_timeout = tonumber(ARGV[3]) " +
"if ctime > 0 then " +
"if redis.call('exists', KEYS[2]) == 1 then " +
"local avg_time = redis.call('get', KEYS[2]) " +
"avg_time = (tonumber(avg_time) * 8 + ctime * 2)/10 " +
"if avg_time >= biz_timeout - 5 then redis.call('set', KEYS[2], avg_time, 'EX', 24*60*60) " +
"else redis.call('del', KEYS[2]) end " +
"elseif ctime > biz_timeout -5 then redis.call('set', KEYS[2], ARGV[2], 'EX', 24*60*60) end " +
"end " +
"return redis.call('del', KEYS[1]) " +
"else return 0 end";
/**
* 续约lua脚本
*/
private static final String RENEW_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], ARGV[2]) else return 0 end";
private final StringRedisTemplate redisTemplate;
public RedisDistributionLockPlus(StringRedisTemplate redisTemplate) {
this.redisTemplate = redisTemplate;
ScheduleTask task = new ScheduleTask(this, lockContentMap);
// 启动定时任务
ScheduleExecutor.schedule(task, 1, 1, TimeUnit.SECONDS);
}
/**
* 加锁
* 取到锁加锁，取不到锁一直等待知道获得锁
*
* @param lockKey
* @param requestId 全局唯一
* @param expire 锁过期时间, 单位秒
* @return
*/
public boolean lock(String lockKey, String requestId, long expire) {
log.info("开始执行加锁, lockKey ={}, requestId={}", lockKey, requestId);
for (; ; ) {
// 判断是否已经有线程持有锁，减少redis的压力
LockContent lockContentOld = lockContentMap.get(lockKey);
boolean unLocked = null == lockContentOld;
// 如果没有被锁，就获取锁
if (unLocked) {
long startTime = System.currentTimeMillis();
// 计算超时时间
long bizExpire = expire == 0L ? DEFAULT_LOCK_TIMEOUT : expire;
String lockKeyRenew = lockKey + "_renew";
RedisScript<Long> script = RedisScript.of(LOCK_SCRIPT, Long.class);
List<String> keys = new ArrayList<>();
keys.add(lockKey);
keys.add(lockKeyRenew);
Long lockExpire = redisTemplate.execute(script, keys, requestId, Long.toString(bizExpire));
if (null != lockExpire && lockExpire > 0) {
// 将锁放入map
LockContent lockContent = new LockContent();
lockContent.setStartTime(startTime);
lockContent.setLockExpire(lockExpire);
lockContent.setExpireTime(startTime + lockExpire * 1000);
lockContent.setRequestId(requestId);
lockContent.setThread(Thread.currentThread());
lockContent.setBizExpire(bizExpire);
lockContent.setLockCount(1);
lockContentMap.put(lockKey, lockContent);
log.info("加锁成功, lockKey ={}, requestId={}", lockKey, requestId);
return true;
}
}
// 重复获取锁，在线程池中由于线程复用，线程相等并不能确定是该线程的锁
if (Thread.currentThread() == lockContentOld.getThread()
&& requestId.equals(lockContentOld.getRequestId())){
// 计数 +1
lockContentOld.setLockCount(lockContentOld.getLockCount()+1);
return true;
}
// 如果被锁或获取锁失败，则等待100毫秒
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
// 这里用lombok 有问题
log.error("获取redis 锁失败, lockKey ={}, requestId={}", lockKey, requestId, e);
return false;
}
}
}
/**
* 解锁
*
* @param lockKey
* @param lockValue
*/
public boolean unlock(String lockKey, String lockValue) {
String lockKeyRenew = lockKey + "_renew";
LockContent lockContent = lockContentMap.get(lockKey);
long consumeTime;
if (null == lockContent) {
consumeTime = 0L;
} else if (lockValue.equals(lockContent.getRequestId())) {
int lockCount = lockContent.getLockCount();
// 每次释放锁， 计数 -1，减到0时删除redis上的key
if (--lockCount > 0) {
lockContent.setLockCount(lockCount);
return false;
}
consumeTime = (System.currentTimeMillis() - lockContent.getStartTime()) / 1000;
} else {
log.info("释放锁失败，不是自己的锁。");
return false;
}
// 删除已完成key，先删除本地缓存，减少redis压力, 分布式锁，只有一个，所以这里不加锁
lockContentMap.remove(lockKey);
RedisScript<Long> script = RedisScript.of(UNLOCK_SCRIPT, Long.class);
List<String> keys = new ArrayList<>();
keys.add(lockKey);
keys.add(lockKeyRenew);
Long result = redisTemplate.execute(script, keys, lockValue, Long.toString(consumeTime),
Long.toString(lockContent.getBizExpire()));
return EXEC_SUCCESS.equals(result);
}
/**
* 续约
*
* @param lockKey
* @param lockContent
* @return true:续约成功，false:续约失败（1、续约期间执行完成，锁被释放 2、不是自己的锁，3、续约期间锁过期了（未解决））
*/
public boolean renew(String lockKey, LockContent lockContent) {
// 检测执行业务线程的状态
Thread.State state = lockContent.getThread().getState();
if (Thread.State.TERMINATED == state) {
log.info("执行业务的线程已终止,不再续约 lockKey ={}, lockContent={}", lockKey, lockContent);
return false;
}
String requestId = lockContent.getRequestId();
long timeOut = (lockContent.getExpireTime() - lockContent.getStartTime()) / 1000;
RedisScript<Long> script = RedisScript.of(RENEW_SCRIPT, Long.class);
List<String> keys = new ArrayList<>();
keys.add(lockKey);
Long result = redisTemplate.execute(script, keys, requestId, Long.toString(timeOut));
log.info("续约结果，True成功，False失败 lockKey ={}, result={}", lockKey, EXEC_SUCCESS.equals(result));
return EXEC_SUCCESS.equals(result);
}
static class ScheduleExecutor {
public static void schedule(ScheduleTask task, long initialDelay, long period, TimeUnit unit) {
long delay = unit.toMillis(initialDelay);
long period_ = unit.toMillis(period);
// 定时执行
new Timer("Lock-Renew-Task").schedule(task, delay, period_);
}
}
static class ScheduleTask extends TimerTask {
private final RedisDistributionLockPlus redisDistributionLock;
private final Map<String, LockContent> lockContentMap;
public ScheduleTask(RedisDistributionLockPlus redisDistributionLock, Map<String, LockContent> lockContentMap) {
this.redisDistributionLock = redisDistributionLock;
this.lockContentMap = lockContentMap;
}
@Override
public void run() {
if (lockContentMap.isEmpty()) {
return;
}
Set<Map.Entry<String, LockContent>> entries = lockContentMap.entrySet();
for (Map.Entry<String, LockContent> entry : entries) {
String lockKey = entry.getKey();
LockContent lockContent = entry.getValue();
long expireTime = lockContent.getExpireTime();
// 减少线程池中任务数量
if ((expireTime - System.currentTimeMillis())/ 1000 < TIME_SECONDS_FIVE) {
//线程池异步续约
ThreadPool.submit(() -> {
boolean renew = redisDistributionLock.renew(lockKey, lockContent);
if (renew) {
long expireTimeNew = lockContent.getStartTime() + (expireTime - lockContent.getStartTime()) * 2 - TIME_SECONDS_FIVE * 1000;
lockContent.setExpireTime(expireTimeNew);
} else {
// 续约失败，说明已经执行完 OR redis 出现问题
lockContentMap.remove(lockKey);
}
});
}
}
}
}
}Salin selepas log masuk
五、redis主从复制的坑
redis高可用最常见的方案就是主从复制（master-slave），这种模式也给redis分布式锁挖了一坑。
redis cluster集群环境下，假如现在A客户端想要加锁，它会根据路由规则选择一台master节点写入key mylock，在加锁成功后，master节点会把key异步复制给对应的slave节点。
如果此时redis master节点宕机，为保证集群可用性，会进行主备切换，slave变为了redis master。A客户端错误地认为它在旧的master节点上成功加锁，但实际上锁已经被B客户端在新的master节点上加上了。
此时就会导致同一时间内多个客户端对一个分布式锁完成了加锁，导致各种脏数据的产生。
至于解决办法嘛，目前看还没有什么根治的方法，只能尽量保证机器的稳定性，减少发生此事件的概率。
小结一下：上面就是我在使用Redis 分布式锁时遇到的一些坑，有点小感慨，经常用一个方法填上这个坑，没多久就发现另一个坑又出来了，其实根本没有什么十全十美的解决方案，哪有什么银弹，只不过是在权衡利弊后，选一个在接受范围内的折中方案而已。
Atas ialah kandungan terperinci Bagaimana untuk memperbaharui kunci yang diedarkan Redis. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!
![[Web front-end] Permulaan pantas Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
