南戴河区网站建设哪家好,中华室内设计协会,一个网站如何进行推广宣传,怎么备案网站空间一、简介
分布式锁其实就是控制分布式系统中不同进程共同访问共享资源的一种锁的实现。在分布式系统中各个微服务都是独立部署在不同的服务器上#xff0c;如果多个服务同时操作同一个共享资源的话#xff0c;就不能像单体服务那样通过synchronized或者Lock等同步机制保证一…一、简介
分布式锁其实就是控制分布式系统中不同进程共同访问共享资源的一种锁的实现。在分布式系统中各个微服务都是独立部署在不同的服务器上如果多个服务同时操作同一个共享资源的话就不能像单体服务那样通过synchronized或者Lock等同步机制保证一个代码块在同一时间只能由一个线程访问来实现共享资源的安全性。因为分布式系统中的不同服务已经不在是多线程之间的并发访问了而是属于多进程之间的并发访问所以就需要一种更加高级的锁机制来处理这种跨JVM进程之间的线程安全问题。
二、主要特征
互斥性在任意时刻只有一个客户端能持有锁。不会发生死锁即使持有锁的客户端发生故障也能保证锁最终会被释放。具有容错性只要大部分的 Redis 节点正常运行客户端就可以加锁和解锁。不乱解锁加锁和解锁必须是同一个客户端客户端自己不能把别人加的锁给解了。
三、实现方案
在 Java 中实现分布式锁的方案有多种常见的3中方案如下
基于数据库实现可以通过数据库的乐观锁或悲观锁实现分布式锁但是由于数据库的 IO 操作比较慢不适合高并发场景而且可能存在死锁和超时等问题。基于 Redis 实现Redis 是一个高性能的内存数据库支持分布式部署可以通过Redis的原子操作实现分布式锁支持key的过期时间设置不容易发生死锁而且具有高性能和高可用性。基于 Lock4j 实现Lock4j 是一个分布式锁组件它提供了多种不同的支持以满足不同性能和环境的需求基于Spring AOP的声明式和编程式分布式锁支持RedisTemplate、Redisson、Zookeeper。基于 ZooKeeper 实现ZooKeeper 是一个高可用性的分布式协调服务可以通过它来实现分布式锁。但是使用 ZooKeeper 需要部署额外的服务增加了系统复杂度。
3.1 基于 Redis 实现 实现方式有以下几种 1setnx expire这种方式加锁操作和设置超时时间是分开的。如果在执行完setnx加锁后正要执行expire设置过期时间时进程挂掉了那这个锁就永远不会过期了。 2set的扩展命令通过set(String key, String value, String nxxx, String expx, int time) 加锁的同时设置过期时间再通过del(key)删除key。这种方式可能导致锁被别的线程误删假设A获取锁后由于业务还没执行完就过期释放了然后立即就被B获取该锁执行业务逻辑此时A执行完成后就会去释放这个锁但此时这个锁已经被B占用了也就是说A此时把B的锁给释放掉了。 3set的扩展命令唯一值校验通过set(String key, String value, String nxxx, String expx, int time) 加锁的同时设置过期时间再通过Lua 脚本去根据唯一值删除key。这种方式可以解决误删除别人的锁问题但是还是存在锁过期释放了业务还没执行完的问题。 添加redis依赖
dependencygroupIdredis.clients/groupIdartifactIdjedis/artifactIdversion2.9.0/version
/dependency创建一个redis配置类用来设置redis连接信息并创建JedisPool和RedisTemplate的实例。
Configuration
public class RedisConfig {Value(${spring.redis.host})private String host;Value(${spring.redis.port})private int port;Value(${spring.redis.timeout})private int timeout;Value(${spring.redis.password})private String password;Value(${spring.redis.pool.maxTotal})private int maxTotal;Value(${spring.redis.pool.maxWait})private int maxWait;Value(${spring.redis.pool.maxIdle})private int maxIdle;Value(${spring.redis.pool.minIdle})private int minIdle;Value(${spring.redis.blockWhenExhausted})private Boolean blockWhenExhausted;Value(${spring.redis.JmxEnabled})private Boolean JmxEnabled;/*** 创建JedisPool实例* * return JedisPool*/RefreshScopeBeanpublic JedisPool jedisPoolFactory() {JedisPoolConfig jedisPoolConfig new JedisPoolConfig();jedisPoolConfig.setMaxTotal(maxTotal);jedisPoolConfig.setMaxIdle(maxIdle);jedisPoolConfig.setMinIdle(minIdle);jedisPoolConfig.setMaxWaitMillis(maxWait);// 连接耗尽时是否阻塞, false报异常,true阻塞直到超时, 默认truejedisPoolConfig.setBlockWhenExhausted(blockWhenExhausted);// 是否启用pool的jmx管理功能, 默认truejedisPoolConfig.setJmxEnabled(JmxEnabled);return new JedisPool(jedisPoolConfig, host, port, timeout, password);}/*** redisTemplate 序列化使用的jdkSerializeable, 存储二进制字节码, 所以自定义序列化类** return redisTemplate*/Beanpublic RedisTemplateString, Object redisTemplate(RedisConnectionFactory redisConnectionFactory) {RedisTemplateString, Object redisTemplate new RedisTemplate();redisTemplate.setConnectionFactory(redisConnectionFactory);// 使用Jackson2JsonRedisSerialize 替换默认序列化SuppressWarnings({ rawtypes, unchecked })Jackson2JsonRedisSerializer jackson2JsonRedisSerializer new Jackson2JsonRedisSerializer(Object.class);ObjectMapper objectMapper new ObjectMapper();objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);jackson2JsonRedisSerializer.setObjectMapper(objectMapper);// 设置value的序列化规则和 key的序列化规则redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);redisTemplate.setHashKeySerializer(new StringRedisSerializer());redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);redisTemplate.afterPropertiesSet();return redisTemplate;}
}编写redis的工具类使用set的扩展命令唯一值校验的方式创建获取和释放分布式锁相应的方法。
Slf4j
Component
public class RedisUtil {Resourceprivate RedisTemplateString, Object redisTemplate;Resourceprivate StringRedisTemplate stringRedisTemplate;private static final String LOCK_SUCCESS OK;/*** NX: 仅在键不存在时设置键* XX: 只有在键已存在时才设置*/private static final String SET_IF_NOT_EXIST NX;/*** 过期时间单位* EX: seconds* PX: milliseconds*/private static final String SET_WITH_EXPIRE_TIME EX;private static final Long RELEASE_SUCCESS 1L;/*** 尝试获取分布式锁** param lockKey 分布式锁的key想要获取锁时判断这个key是否存在于redis中存在则说明获取分布式锁失败否则成功获取锁* param requestId 每个请求的全局唯一id用于释放锁时只能释放自己持有的锁* param expireTime 超期时间,单位秒* return boolean 是否获取成功*/public boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {if (jedis null) {return false;}String result jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);if (LOCK_SUCCESS.equals(result)) {log.info(获取分布式锁成功, lockKey is{}, requestId is:{}, lockKey, requestId);return true;}log.info(获取分布式锁失败, lockKey is{}, requestId is:{}, lockKey, requestId);return false;}/*** 释放分布式锁** param jedis Redis客户端* param lockKey 分布式锁的key* param requestId 每个请求的全局唯一id* return 是否释放成功*/public boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {if (jedis null) {log.info(分布式锁释放失败,Jedis为空, lockKey is{}, requestId is:{}, lockKey, requestId);return false;}// 通过Lua 脚本保证只释放requestId对应的lockKeyString script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end;Object result jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));if (RELEASE_SUCCESS.equals(result)) {log.info(分布式锁释放成功, lockKey is{}, requestId is:{}, lockKey, requestId);return true;}log.info(分布式锁释放失败, lockKey is{}, requestId is:{}, lockKey, requestId);return false;}
}案例实现 /*** 通过分布式锁来生成全局订单唯一的id* * param type 订单类型* return String 订单唯一的id*/Overridepublic String generate(String type) {String orderId null;log.info(要生成的订单类型为{}, type);if (StringUtils.isBlank(type)) {return null;}// 开始获取分布式锁Jedis jedis jedisPool.getResource();String lockKey redisUtils.getRedisKey(RedisTemplateConstant.DISTRIBUTED_LOCK_KEY_TYPE, type);String requestId CommonUtil.getUUID();try {if (redisUtils.tryGetDistributedLock(jedis, lockKey, requestId, expireTime)) {// 生成订单idorderId getOrderId(type);}} catch (Exception e) {log.info(组装id出错);} finally {// 释放分布式锁redisUtils.releaseDistributedLock(jedis, lockKey, requestId);if (jedis ! null) {jedis.close();}}return orderId;}3.2 基于 Redisson RedLock
Redisson 对于可能存在锁过期释放业务没执行完的问题。Redisson的处理逻辑是只要客户端一加锁成功就会启动一个watch dog看门狗它是一个后台线程会每隔10秒检查一下如果当前线程还持有锁那么就会不断的延长锁key的生存时间。因此Redisson就是使用 watch dog 解决了锁过期释放业务没执行完问题。RedLock 由于Redis一般都是集群部署如果客户端A在Redis的master节点上拿到了锁但是加锁的key还没同步到slave节点。恰好这时master节点发生故障一个slave节点就会升级为master节点。客户端B就可以顺理成章获取同个key的锁了但客户端A早在故障的master节点已经拿到这个锁了从而导致分布式锁在这种情况下失效了。因此可通过 redLock 红锁一种相对安全的分布式锁实现方式采用主节点过半机制当客户端申请分布式锁的时候需要向所有的redis实例发出申请只有超过半数的redis实例报告获取锁成功才能算真正获取到锁。也因此redis集群部署的节点数一般都为奇数。RedLock 已经在最新本的 Redisson 中被弃用了 其实红锁其实也并不能解决根本问题只是降低问题发生的概率。因为完全相互独立的redis每一台至少也要保证高可用还是会有主从节点。既然有主从节点在持续的高并发下master还是可能会宕机从节点可能还没来得及同步锁的数据。在实际场景中红锁是很少使用的。这是因为使用了红锁后会影响高并发环境下的性能使得程序的体验更差。同时使用红锁后当加锁成功的RLock个数不超过总数的一半时会返回加锁失败即使在业务层面任务加锁成功了但是红锁也会返回加锁失败的结果。另外实现红锁需要提供多套Redis的主从部署架构并且这多套Redis主从架构中的Master节点必须都是独立的相互之间没有任何数据交互。引入依赖
dependencygroupIdorg.redisson/groupIdartifactIdredisson-spring-boot-starter/artifactIdversion3.20.0/version
/dependency编写分布式锁工具类
Slf4j
Component
public class RedissonUtil {Resourceprivate Redisson redisson;/*** 加锁** param key 分布式锁的 key* param timeout 超时时间* param unit 时间单位* return*/public boolean tryLock(String key, long timeout, TimeUnit unit) {RLock lock redisson.getLock(key);try {return lock.tryLock(timeout, unit);} catch (InterruptedException e) {Thread.currentThread().interrupt();return false;}}/*** 释放分布式锁** param key 分布式锁的 key*/public void unlock(String key) {RLock lock redisson.getLock(key);lock.unlock();}
}3.3 基于 Lock4j 实现
引入依赖 !-- 若使用redisTemplate作为分布式锁底层则需要引入 --
dependencygroupIdcom.baomidou/groupIdartifactIdlock4j-redis-template-spring-boot-starter/artifactIdversion2.2.4/version
/dependency
!-- 若使用redisson作为分布式锁底层则需要引入 --
dependencygroupIdcom.baomidou/groupIdartifactIdlock4j-redisson-spring-boot-starter/artifactIdversion2.2.4/version
/dependency简单使用实例
Slf4j
RestController
RequestMapping(/redisLock)
public class RedisLockController {Autowiredprivate LockTemplate lockTemplate;/*** 使用 lock4j 注解加锁*/Lock4j(keys {#key}, acquireTimeout 1000, expire 10000)GetMapping(/testAnnotate)public RString testAnnotate(String key) {try {Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}return R.ok(key);}/*** 使用LockTemplate模板加锁*/GetMapping(/testLock4jLockTemplate)public RString testLock4jLockTemplate(String key) {final LockInfo lockInfo lockTemplate.lock(key, 30000L, 5000L, RedissonLockExecutor.class);if (null lockInfo) {throw new RuntimeException(业务繁忙,请稍后再试!);}// 获取锁成功处理业务try {try {Thread.sleep(8000);} catch (InterruptedException e) {//}System.out.println(当前线程: Thread.currentThread().getName());} finally {//释放锁lockTemplate.releaseLock(lockInfo);}return R.ok(key);}
}3.4 基于 ZooKeeper 实现
引入依赖
dependencygroupIdorg.apache.curator/groupIdartifactIdcurator-framework/artifactIdversionlatest/version
/dependency
dependencygroupIdorg.apache.curator/groupIdartifactIdcurator-recipes/artifactIdversionlatest/version
/dependency
dependencygroupIdorg.apache.zookeeper/groupIdartifactIdzookeeper/artifactIdversionlatest/version
/dependency配置 ZooKeeper 连接
spring:zookeeper:connect-string: localhost:8081namespace: test编写分布式锁工具类
Component
public class ZkLockUtil {Autowiredprivate CuratorFramework curatorFramework;/*** 获取分布式锁** param lockPath 锁路径* param waitTime 等待时间* param leaseTime 锁持有时间* param timeUnit 时间单位* return 锁对象* throws Exception 获取锁异常*/public InterProcessMutex acquire(String lockPath, long waitTime, long leaseTime, TimeUnit timeUnit) throws Exception {InterProcessMutex lock new InterProcessMutex(curatorFramework, lockPath);if (!lock.acquire(waitTime, timeUnit)) {throw new RuntimeException(获取分布式锁失败);}if (leaseTime 0) {lock.acquire(leaseTime, timeUnit);}return lock;}/*** 释放分布式锁** param lock 锁对象* throws Exception 释放锁异常*/public void release(InterProcessMutex lock) throws Exception {if (lock ! null) {lock.release();}}
}
