哪哪个网站可以做兼职,wordpress怎么发文章,免费建站网站制作模板,以公司做网站Redis 分布式锁如何自动续期
何为分布式
分布式#xff0c;从狭义上理解#xff0c;也与集群差不多#xff0c;但是它的组织比较松散#xff0c;不像集群#xff0c;有一定组织性#xff0c;一台服务器宕了#xff0c;其他的服务器可以顶上来。分布式的每一个节点从狭义上理解也与集群差不多但是它的组织比较松散不像集群有一定组织性一台服务器宕了其他的服务器可以顶上来。分布式的每一个节点都完成不同的业务一个节点宕了这个业务就不可访问了。分布式是指将一个业务拆分不同的子业务分布在不同的机器上执行。
分布式锁
为了保证操作共享资源在高并发情况下的同一时间只能被同一个线程执行在单体应用单机部署的情况下可以使用Java并发处理相关的API(如ReentrantLcok或synchronized)进行互斥控制这是在JVM层面的加锁方式。单体单机部署的系统被演化成分布式系统后由于分布式系统多线程、多进程并且分布在不同机器上这将使原单机部署情况下的并发控制锁策略失效为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问这就是分布式锁要解决的问题 分布式锁是一种用于在分布式系统中实现互斥访问的机制。它可以确保在多个节点同时访问共享资源时只有一个节点能够获取到锁并执行操作其他节点需要等待。
分布式锁的实现方式
基于数据库
可以使用数据库的事务机制来实现分布式锁。通过在数据库中创建一个特定的表或记录来表示锁的状态当节点需要获取锁时尝试插入或更新这个表或记录如果成功则获取到锁否则等待。
基于缓存
可以使用分布式缓存如Redis或Memcached来实现分布式锁。通过在缓存中设置一个特定的键值对来表示锁的状态当节点需要获取锁时尝试设置这个键值对如果成功则获取到锁否则等待。
基于ZooKeeper
ZooKeeper是一个分布式协调服务可以用于实现分布式锁。通过创建临时顺序节点来表示锁的状态当节点需要获取锁时尝试创建自己的临时顺序节点并检查是否是最小的节点如果是则获取到锁否则监听前一个节点的删除事件等待。
基于分布式算法
还有一些基于分布式算法的实现方式如Chubby、Raft等。这些算法通过选举、协调等机制来实现分布式锁。 需要注意的是分布式锁的实现需要考虑到并发性、可靠性和性能等方面的问题选择合适的实现方式需要根据具体的需求和场景进行评估。 分布式锁的特点
互斥性在任意时刻只有一个客户端能持有锁。不会发生死锁即使有一个客户端在持有锁的期间崩溃而没有主动解锁也能保证后续其他客户端能加锁。容错性只要大部分的Redis节点正常运行客户端就可以加锁和解锁。可重入性加锁和解锁必须是同一个客户端客户端自己不能把别人加的锁给解了。
Redis实现分布式锁
Redis Setnx命令
Redis SetnxSET if Not eXists 命令在指定的 key 不存在时为 key 设置指定的值。
redis 127.0.0.1:6379 SETNX KEY_NAME VALUE设置成功返回 1 。 设置失败返回 0 。
Set命令 setnx不能同时完成expire设置失效时长不能保证setnx和expire的原子性。我们可以使用set命令完成setnx和expire的操作并且这种操作是原子操作。 例子设置locktest失效时长3s不存在时设置 set lock test ex 3 nx。设置成功返回OK设置失败返回null
SpringBoot使用Redis分布式锁
基于RedisTemplate
假设业务代码块在 6s之内处理完成那么下面的代码就不会有业务代码执行超时分布式锁没有问题如果业务代码执行耗时较长那么设置的键会自动过期导致上个业务还没有执行结束下个业务还能拿到锁分布式锁失效
/*** Set 实现分布式锁子*/
Override
public void setRedisLock() {// redis KeyString redisKey ID_1001;// value 身份标识String redisValue UUID.randomUUID().toString();try {// 获取分布式锁设置超时时间 6s 假设业务代码最长 6s 执行完毕ValueOperations valueOperations redisTemplate.opsForValue();boolean lockFlag !valueOperations.setIfAbsent(redisKey, redisValue, 6, TimeUnit.SECONDS).booleanValue();if (lockFlag) {throw new Exception(redis key redisKey 值 redisValue 获取锁失败);} else {logger.info(redis key{} 值{} 获取锁成功, redisKey, redisValue);}// 实现业务代码暂时假设业务代码执行时长在 6s 之内} catch (Exception e) {logger.error(e.getMessage(), e);throw new RuntimeException(e.getMessage());} finally {boolean deleteFlag;String currentValue (String) redisTemplate.opsForValue().get(redisKey);if (redisValue.equals(currentValue)) {deleteFlag redisTemplate.opsForValue().getOperations().delete(redisKey).booleanValue();if (deleteFlag) {logger.info(redis 锁{} 释放成功, redisKey);} else {logger.error(redis 锁{} 释放失败, redisKey);}} else {logger.error(redis 锁{} 值{} 身份校验失败无法释放, redisKey, redisValue);}}}Redis分布式锁续期处理
在上面的例子中当业务代码执行耗时超过redis设置的超时时间时下一个任务获取锁的时候还是会获取成功这样在业务上是又问题的。所以得要考虑处理锁续期。实现思路开启一个定时任务作为守护线程如果业务代码没有执行完成主动进行续期操作任务完整之后终止守护线程释放获取的锁
Override
public void setRedisLock1() {// redis KeyString redisKey ID_1001;TestTask testTask new TestTask();CustomResponse response execute(testTask, redisKey, 7, true);if (response.getCode() ! 0) {logger.error(线程 Thread.currentThread().getId() 执行结果 response.getMsg());}
}/*** 利用redis做分布式锁** param runnable 执行的业务* param lockKey 锁定key 不同业务应该全局唯一* param lockTime 锁定时间 单位 ms* param autoRelock 是否自动续期*/
public CustomResponse execute(Runnable runnable, String lockKey, long lockTime, boolean autoRelock) {CustomResponse customResponse new CustomResponse();execute(runnable, lockKey, lockTime, autoRelock, customResponse);return customResponse;
}/*** 利用redis做分布式锁** param runnable 执行的业务* param lockKey 锁定key 不同业务应该全局唯一* param lockTime 锁定时间 单位 ms* param autoRelock 是否自动续期* param customResponse 执行结果*/
public void execute(Runnable runnable, String lockKey, long lockTime, boolean autoRelock, CustomResponse customResponse) {if (customResponse null) {throw new IllegalArgumentException(customResponse 参数不能为空);}if (lockTime 0) {throw new IllegalArgumentException(请设置正确的 redis key 超时时间);}boolean flag true;boolean completedFlag true;TimerTask timerTask null;ScheduledFuture? scheduledFuture null;try {// 失效时间设置失败的key强制删除Long hasKeyExpire redisTemplate.getExpire(lockKey);if (hasKeyExpire ! null hasKeyExpire.intValue() -1) {redisTemplate.delete(lockKey);}ValueOperationsString, String operations redisTemplate.opsForValue();if (Boolean.TRUE.equals(operations.setIfAbsent(lockKey, 1, lockTime, TimeUnit.MILLISECONDS))) {// 开启续期超时时间之后开始任务if (autoRelock) {timerTask new TimerTask() {public void run() {logger.info(redis key:{} 自动续期任务执行..., lockKey);redisTemplate.opsForValue().setIfPresent(lockKey, 1, lockTime, TimeUnit.MILLISECONDS);}};try {scheduledFuture scheduledExecutorService.scheduleAtFixedRate(timerTask, lockTime / 2, lockTime, TimeUnit.SECONDS);} catch (Throwable e) {logger.debug(e.getMessage());}}customResponse.setMsg(0, 获取 redis 锁成功);// 执行业务逻辑try {runnable.run();// 处理标志位completedFlag false;} catch (Throwable e) {logger.error(redis key{} 执行业务代码出错{}, lockKey, e.getMessage(), e);customResponse.setMsg(500, e.getMessage());}} else {flag false;customResponse.setMsg(100, 获取锁失败);}} catch (Throwable e) {if (completedFlag) {logger.error(e.getMessage(), e);customResponse.setMsg(500, e.getMessage());}} finally {try {// 删除自己设置的锁if (flag) {redisTemplate.delete(lockKey);logger.info(执行完成删除自己的 key);}// 移除定时任务timerTask.cancel();if (Objects.nonNull(scheduledFuture)) {scheduledFuture.cancel(true);}} catch (Throwable e) {logger.debug(e.getMessage(), e);}}
}private class TestTask implements Runnable {Overridepublic void run() {try {logger.info(任务开始执行...);Thread.sleep(10000);logger.info(任务执行结束...);} catch (InterruptedException e) {throw new RuntimeException(e);}}}获取锁使用Redis的SETNX命令尝试获取锁。如果返回1表示获取锁成功返回0表示锁已被其他进程持有。设置锁的过期时间如果成功获取到锁可以使用Redis的EXPIRE命令设置锁的过期时间确保在一定时间后自动释放锁。续期处理在业务处理过程中可以定期比如锁过期时间的一半使用Redis的EXPIRE命令来延长锁的过期时间防止锁过期后被其他进程获取。释放锁在业务处理完成后使用Redis的DEL命令释放锁。需要注意的是分布式锁的续期处理需要保证原子性避免多个进程同时续期导致锁被误释放。可以使用Redis的Lua脚本来保证续期操作的原子性。 另外为了防止进程异常退出或崩溃导致锁无法释放可以使用Redis的SET命令设置一个唯一的锁标识并在获取锁和续期操作时进行比对确保只有持有锁的进程才能释放锁。
