集群分布式场景高并发
1.negix配置代理和路由
高并发场景超卖问题
1.使用原生redis控制超卖时(若是商品，则可以将商品id作为锁对象)，会遇到的问题
问题一：若直接使用：将获取锁的对象和设置的超时的时间分开，则不能控制原子性，如下所示
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "zhuge");
stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);
问题二：若直接使用：将获取锁的对象和设置的超时的时间放在一个原子操作里执行时，在临界条件下，当程序执行到最后准备释放锁时候，锁的超时时间已到，则此时的锁成为已过期，则释放不了锁；而当下一个线程也来执行任务时，前一个任务将这个任务所拿的所给释放掉了（释放掉不属于自己的锁对象）；则引入redisson分布式锁来解决当前的问题，redisson具有锁续命机制
@RestController
public class IndexController {
@Autowired
private Redisson redisson;
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private 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包
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.6.5</version>
</dependency>
设置redission配置
@SpringBootApplication
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
@Bean
public Redisson redisson() {
// 此为单机模式
Config config = new Config();
config.useSingleServer().setAddress("redis://localhost:6379").setDatabase(0);
return (Redisson) Redisson.create(config);
}
}redission的基本使用
@RestController
public class IndexController {
@Autowired
private Redisson redisson;
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private 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(); 加锁
@Override
public void lockInterruptibly() throws InterruptedException {
lockInterruptibly(-1, null);
}执行 lockInterruptibly加锁逻辑
@Override
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(leaseTime, unit, threadId); //尝试去加锁，返回的时加锁后的过期时间
// lock acquired
if (ttl == null) { //若ttl为null ,则表示加锁成功 ;若ttl不为null ,则往下走
return;
}
RFuture<RedissonLockEntry> future = subscribe(threadId); //发布订阅，订阅前者执行的任务若提前执行完，则唤醒机制，去重新获取锁
commandExecutor.syncSubscription(future);
try {
while (true) { //进入循环
ttl = tryAcquire(leaseTime, unit, threadId); //再次尝试获取锁，返回加锁成功后的过期时间
// lock acquired
if (ttl == null) { //若ttl为null ,则表示加锁成功 ;若ttl不为null ,则往下走
break;
}
// waiting for message
if (ttl >= 0) { 若ttl大于o
getEntry(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> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
if (leaseTime != -1) {
return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}
//leaseTime 默认设置为-1
RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
//执行加锁逻辑
ttlRemainingFuture.addListener(new FutureListener<Long>() { //异步执行，锁续命逻辑
@Override
public void operationComplete(Future<Long> future) throws Exception {
if (!future.isSuccess()) { //若加锁不成功，则退出
return;
}
Long ttlRemaining = future.getNow(); //若加锁成功，则ttlRemaining 为null
// lock acquired
if (ttlRemaining == null) {
scheduleExpirationRenewal(threadId); //加锁成功，则执行锁续命逻辑
}
}
});
return ttlRemainingFuture;
}执行tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); 加锁逻辑
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> 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]表示下面的internalLockLeaseTime
"return 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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}
执行scheduleExpirationRenewal(threadId);锁续命逻辑
private void scheduleExpirationRenewal(final long threadId) {
if (expirationRenewalMap.containsKey(getEntryName())) {
return;
}
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
//执行锁续命逻辑
RFuture<Boolean> 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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
//异步监听执行
future.addListener(new FutureListener<Boolean>() {
@Override
public void operationComplete(Future<Boolean> future) throws Exception {
expirationRenewalMap.remove(getEntryName());
if (!future.isSuccess()) {
log.error("Can't update lock " + getName() + " expiration", future.cause());
return;
}
//判断是否任然持有锁，是的话，则getNow（）为null
if (future.getNow()) {
// reschedule itself
scheduleExpirationRenewal(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);
