分布式锁实现方案-Redis
一.Redis分布式锁实现分析
存在的问题：
如果某个线程执行的太慢，导致在有效期内还没有执行完，那么因为设置了锁超时自动释放机制，此时锁被自动释放，另一个线程进来拿到锁开始执行代码，就会出现同一时间有两个线程在执行互斥资源代码，可能出现数据不一致。
如何解决：
设置合理的超时时间 + 监控代码执行情况
自动续期，起一个定时任务，周期性扫描超距离时时间还剩多少时仍没有执行完的线程，自动延长时间。注定只能使用其他机制解决的难题
二.分布式锁实现方案
Redis锁开源实现：
三.自己实现一个基于Redis的分布式锁
package com.hong.redis;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
- @author wanghong
**/
public class RedisLockImpl implements Lock {
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 你要锁的资源
*/
private String resourceName;
int timeout;
public RedisLockImpl(StringRedisTemplate stringRedisTemplate, String
resourceName, int timeout) {
this.stringRedisTemplate = stringRedisTemplate;
this.resourceName = "lock_" + resourceName;
this.timeout = timeout;
}
private Lock lock = new ReentrantLock();
/**
* 一直要等到抢到锁为止
*/
@Override
public void lock() {
// 限制同一个JVM进程内的资源竞争,分布式锁（JVM之间的竞争） + JVM锁
lock.lock();
try {
while (!tryLock()) {
// 订阅指定的Redis主题，接收释放锁的信号
stringRedisTemplate.execute(new RedisCallback<Long>() {
@Override
public Long doInRedis(RedisConnection redisConnection) throws
DataAccessException {
try {
CountDownLatch latch = new CountDownLatch(1);
// subscribe立马返回，是否订阅完毕，异步触发
redisConnection.subscribe((message, pattern) -> {
// 收到消息，不管结果，立刻再次抢锁
latch.countDown();
}, ("release_lock_" + resourceName).getBytes());
//等待有通知，才继续循环
latch.await(timeout, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 0L;
}
});
}
} finally {
lock.unlock();
}
}
@Override
public void lockInterruptibly() throws InterruptedException { }
@Override
public boolean tryLock() {
// set命令，往redis存放锁的标记
Boolean lockResult = stringRedisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {
String value = "";
Boolean result = redisConnection.set(resourceName.getBytes(),
value.getBytes(),
Expiration.seconds(timeout),
RedisStringCommands.SetOption.SET_IF_ABSENT
);
return result;
}
});
return lockResult;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return false;
}
@Override
public void unlock() {
stringRedisTemplate.delete(resourceName);
//通过Redis发布订阅机制，发送一个通知给其他等待的请求
stringRedisTemplate.execute(new RedisCallback<Long>() {
@Override
public Long doInRedis(RedisConnection redisConnection) throws DataAccessException {
return redisConnection.publish(("release_lock_" + resourceName).getBytes(), resourceName.getBytes());
}
});
}
@Override
public Condition newCondition() {
return null;
}
}
