天猫旗舰店网站建设案例,软文营销的方法,前端网站大全,wordpress邮件key集群分布式场景高并发
1.negix配置代理和路由 高并发场景超卖问题
1.使用原生redis控制超卖时(若是商品#xff0c;则可以将商品id作为锁对象)#xff0c;会遇到的问题
问题一#xff1a;若直接使用#xff1a;将获取锁的对象和设置的超时的时间分开#xff0c;则不能控…集群分布式场景高并发
1.negix配置代理和路由 高并发场景超卖问题
1.使用原生redis控制超卖时(若是商品则可以将商品id作为锁对象)会遇到的问题
问题一若直接使用将获取锁的对象和设置的超时的时间分开则不能控制原子性如下所示 Boolean result stringRedisTemplate.opsForValue().setIfAbsent(lockKey, zhuge); stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS); 问题二若直接使用将获取锁的对象和设置的超时的时间放在一个原子操作里执行时在临界条件下当程序执行到最后准备释放锁时候锁的超时时间已到则此时的锁成为已过期则释放不了锁而当下一个线程也来执行任务时前一个任务将这个任务所拿的所给释放掉了释放掉不属于自己的锁对象则引入redisson分布式锁来解决当前的问题redisson具有锁续命机制
RestController
public class IndexController {Autowiredprivate Redisson redisson;Autowiredprivate StringRedisTemplate stringRedisTemplate;Autowiredprivate RedisTemplate redisTemplate;RequestMapping(/deduct_stock)public String deductStock() {String lockKey lock:product_101;//Boolean result stringRedisTemplate.opsForValue().setIfAbsent(lockKey, zhuge);//stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);String clientId UUID.randomUUID().toString();Boolean result stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS); //jedis.setnx(k,v)if (!result) {return error_code;}try {int stock Integer.parseInt(stringRedisTemplate.opsForValue().get(stock)); // jedis.get(stock)if (stock 0) {int realStock stock - 1;stringRedisTemplate.opsForValue().set(stock, realStock ); // jedis.set(key,value)System.out.println(扣减成功剩余库存: realStock);} else {System.out.println(扣减失败库存不足);}} finally {if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {stringRedisTemplate.delete(lockKey);}}return end;}
使用分布式锁redisson
redisson使用
引入对应的redission的jar包 dependencygroupIdorg.redisson/groupIdartifactIdredisson/artifactIdversion3.6.5/version/dependency设置redission配置 SpringBootApplication
public class Application {public static void main(String[] args) {SpringApplication.run(Application.class, args);}Beanpublic Redisson redisson() {// 此为单机模式Config config new Config();config.useSingleServer().setAddress(redis://localhost:6379).setDatabase(0);return (Redisson) Redisson.create(config);}} redission的基本使用 RestController
public class IndexController {Autowiredprivate Redisson redisson;Autowiredprivate StringRedisTemplate stringRedisTemplate;Autowiredprivate RedisTemplate redisTemplate;RequestMapping(/deduct_stock)public String deductStock() {String lockKey lock:product_101;//Boolean result stringRedisTemplate.opsForValue().setIfAbsent(lockKey, zhuge);//stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);/*String clientId UUID.randomUUID().toString();Boolean result stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS); //jedis.setnx(k,v)if (!result) {return error_code;}*///获取锁对象RLock redissonLock redisson.getLock(lockKey);//加分布式锁redissonLock.lock(); // .setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);try {int stock Integer.parseInt(stringRedisTemplate.opsForValue().get(stock)); // jedis.get(stock)if (stock 0) {int realStock stock - 1;stringRedisTemplate.opsForValue().set(stock, realStock ); // jedis.set(key,value)System.out.println(扣减成功剩余库存: realStock);} else {System.out.println(扣减失败库存不足);}} finally {/*if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {stringRedisTemplate.delete(lockKey);}*///解锁redissonLock.unlock();}return end;}
Redission执行的逻辑流程 Redission分布式锁加锁源码分析 redissonLock.lock();
加锁 Overridepublic void lockInterruptibly() throws InterruptedException {lockInterruptibly(-1, null);}
执行 lockInterruptibly加锁逻辑
Overridepublic void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {long threadId Thread.currentThread().getId();Long ttl tryAcquire(leaseTime, unit, threadId); //尝试去加锁返回的时加锁后的过期时间// lock acquiredif (ttl null) { //若ttl为null ,则表示加锁成功 ;若ttl不为null ,则往下走return;}RFutureRedissonLockEntry future subscribe(threadId); //发布订阅订阅前者执行的任务若提前执行完则唤醒机制去重新获取锁commandExecutor.syncSubscription(future);try {while (true) { //进入循环ttl tryAcquire(leaseTime, unit, threadId); //再次尝试获取锁返回加锁成功后的过期时间// lock acquiredif (ttl null) { //若ttl为null ,则表示加锁成功 ;若ttl不为null ,则往下走break;}// waiting for messageif (ttl 0) { 若ttl大于ogetEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
//进行间歇性锁自旋逻辑不占用cpu资源} else {getEntry(threadId).getLatch().acquire();}}} finally {unsubscribe(future, threadId);}
// get(lockAsync(leaseTime, unit));}
执行tryAcquire(leaseTime, unit, threadId)尝试加锁逻辑 private T RFutureLong tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {if (leaseTime ! -1) {return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);}
//leaseTime 默认设置为-1RFutureLong ttlRemainingFuture tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
//执行加锁逻辑ttlRemainingFuture.addListener(new FutureListenerLong() { //异步执行锁续命逻辑Overridepublic void operationComplete(FutureLong future) throws Exception {if (!future.isSuccess()) { //若加锁不成功则退出return;}Long ttlRemaining future.getNow(); //若加锁成功则ttlRemaining 为null// lock acquiredif (ttlRemaining null) {scheduleExpirationRenewal(threadId); //加锁成功则执行锁续命逻辑}}});return ttlRemainingFuture;}
执行tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); 加锁逻辑 T RFutureT tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommandT command) {
//通过lua脚本来执行加锁逻辑来保证原子性internalLockLeaseTime unit.toMillis(leaseTime);return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,//执行加锁逻辑 key,argv与下面所传参数一一对应if (redis.call(exists, KEYS[1]) 0) then //KEY[1]表示下面的getName(),redis.call(hset, KEYS[1], ARGV[2], 1); //ARGV[2]表示下面的getLockName(threadId)redis.call(pexpire, KEYS[1], ARGV[1]); //ARGV[1]表示下面的internalLockLeaseTimereturn nil; end;
//执行锁重入逻辑一个线程对同一个锁对象进行多次加锁此为重入锁逻辑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; return redis.call(pttl, KEYS[1]);, //设置返回过期时间Collections.ObjectsingletonList(getName()), internalLockLeaseTime, getLockName(threadId));}
执行scheduleExpirationRenewal(threadId);锁续命逻辑
private void scheduleExpirationRenewal(final long threadId) {if (expirationRenewalMap.containsKey(getEntryName())) {return;}Timeout task commandExecutor.getConnectionManager().newTimeout(new TimerTask() {Overridepublic void run(Timeout timeout) throws Exception {//执行锁续命逻辑RFutureBoolean future commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
//KEYS[1]表示getName()ARGV[2]表示锁对象getLockName(threadId) ARGV[1]表示过期时间
//判断当前锁的对象为当前的线程对象那么则当前的锁的对象设置原始的过期时间以达到续命效果if (redis.call(hexists, KEYS[1], ARGV[2]) 1) then redis.call(pexpire, KEYS[1], ARGV[1]); return 1; end; return 0;,Collections.ObjectsingletonList(getName()), internalLockLeaseTime, getLockName(threadId));//异步监听执行 future.addListener(new FutureListenerBoolean() {Overridepublic void operationComplete(FutureBoolean future) throws Exception {expirationRenewalMap.remove(getEntryName());if (!future.isSuccess()) {log.error(Cant update lock getName() expiration, future.cause());return;}//判断是否任然持有锁是的话则getNow为nullif (future.getNow()) {// reschedule itselfscheduleExpirationRenewal(threadId);//再次执行续命逻辑}}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);if (expirationRenewalMap.putIfAbsent(getEntryName(), task) ! null) {task.cancel();}}
在外层未获取到锁的线程 getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);间隙自旋获取锁对象
getLatch()表示信号量信号量为1则表示阻塞状态最终通过发布订阅方式来唤醒当前被阻塞的线程唤醒后则执行获取锁的逻辑 doAcquireSharedNanos(arg, nanosTimeout);
