分布式锁
2019-06-24 本文已影响0人
神豪VS勇士赢
方案一：
redis
方案二：
zk
方案一：redis
public class RedisTool {
private static final String LOCK_SUCCESS = "OK";
private static final String SET_IF_NOT_EXIST = "NX";
private static final String SET_WITH_EXPIRE_TIME = "PX";
private static final Long RELEASE_SUCCESS = 1L;
private static int ticketCount = 450;
public static JedisPool pool;
static String lockKey = getRequestId();
static {
if (pool == null) {
JedisPoolConfig config = new JedisPoolConfig();
//控制一个pool可分配多少个jedis实例，通过pool.getResource()来获取；
//如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis实例，则此时pool的状态为exhausted(耗尽)。
config.setMaxTotal(1000);
config.setMaxWaitMillis(1000);
//控制一个pool最多有多少个状态为idle(空闲的)的jedis实例。
config.setMaxIdle(50);
//表示当borrow(引入)一个jedis实例时，最大的等待时间，如果超过等待时间，则直接抛出JedisConnectionException；单位毫秒
//小于零:阻塞不确定的时间, 默认-1
config.setMaxWaitMillis(1000 * 100);
//在borrow(引入)一个jedis实例时，是否提前进行validate操作；如果为true，则得到的jedis实例均是可用的；
config.setTestOnBorrow(true);
//return 一个jedis实例给pool时，是否检查连接可用性（ping()）
config.setTestOnReturn(true);
//connectionTimeout 连接超时（默认2000ms）
//soTimeout 响应超时（默认2000ms）
pool = new JedisPool(config, "127.0.0.1", 6379, 2000);
}
}
public static Jedis getJedis() {
return pool.getResource();
}
public static String getRequestId() {
String uuid = UUID.randomUUID().toString().replace("-", "");
return uuid;
}
@Test
public void test() throws InterruptedException {
long start = System.currentTimeMillis();
final CountDownLatch countDownLatch = new CountDownLatch(500);
Executor executor = Executors.newFixedThreadPool(50);
for (int i = 0; i < 500; i++) {
executor.execute(new Runnable() {
@Override
public void run() {
try {
distributeLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
countDownLatch.countDown();
}
});
}
countDownLatch.await();
long costTime = System.currentTimeMillis() - start;
pool.close();
System.out.println("it totally took：" + costTime + " ms");
}
public void distributeLock() throws InterruptedException {
Jedis jedis = null;
String requestId = null;
int timeoutCount = 0;
requestId = getRequestId();
while (true) {
try {
jedis = getJedis();
break;
} catch (Exception e) {
if (e instanceof JedisConnectionException || e instanceof SocketTimeoutException) {
//记录下获取多少次才获得jedis连接，并发很多的时候可能池里的连接不够而导致获取连接失败，所以这里循环获取
timeoutCount++;
//System.out.println("user:"+requestId+" getJedis timeoutCount={"+timeoutCount+"}");
}
}
}
if (tryGetDistributedLock(jedis, lockKey, requestId, 3000)) {
if (ticketCount > 0) {
System.out.println(requestId + " have got a ticket：" + ticketCount);
ticketCount--;
} else {
System.out.println(requestId + "the ticket have been sold out.");
}
System.out.println(requestId+"释放了锁");
releaseDistributedLock(jedis, lockKey, requestId);
jedis.close();
} else {
System.out.println("user" + requestId + " have been sold out!Give up getting lock");
}
}
/**
* 尝试获取分布式锁
*
* @param jedis Redis客户端
* @param lockKey 锁
* @param requestId 请求标识
* @param expireTime 超期时间
* @return 是否获取成功
*/
public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) throws InterruptedException {
int timeoutCount = 0;
long currentTime = System.currentTimeMillis();
while (true) {
try {
String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
if (ticketCount <= 0) {
System.out.println("user" + requestId + " have been sold out!Give up getting lock");
return false;
}
/*//等待了60S以上，直接不再获取锁
if(System.currentTimeMillis() > (currentTime+ 60*1000)){
System.out.println("user"+ requestId +"等待了60S以上了，放弃！！！");
return false;
}*/
} catch (Exception e) {
/*if (e instanceof JedisConnectionException || e instanceof SocketTimeoutException) {
timeoutCount++;
//System.out.println("user:"+requestId+" jedis.set() timeoutCount={"+timeoutCount+"}");
if (timeoutCount > 3)
{
System.out.println("connect error！");
break;
}
}*/
}
}
}
/**
* 释放分布式锁
*
* @param jedis Redis客户端
* @param lockKey 锁
* @param requestId 请求标识
* @return 是否释放成功
*/
public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
//判断requestId相等时为了确定解锁和获取锁的用户是同一个用户
//lockKey是针对此业务的锁ID 执行成功返回1.
String 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)) {
return true;
}
return false;
}
}
方案二：zk
public class LockByCurator {
private static Integer a = 0;
private static String CONNECT_SERVER = "127.0.0.1";
private static int SESSION_TIMEOUT = 3000;
private static int CONNECTION_TIMEOUT = 3000;
private static final String CURATOR_LOCK = "/currentLock";
private static void doLock(CuratorFramework cf) {
System.out.println(Thread.currentThread().getName() + "尝试获取锁");
InterProcessMutex mutex = new InterProcessMutex(cf, CURATOR_LOCK);
try {
if (mutex.acquire(5, TimeUnit.SECONDS)) {
System.out.println(Thread.currentThread().getName() + "获得到了锁");
a++;
System.out.println("=====================>"+a);
/*操作业务*/
Thread.sleep(5000);
}
} catch (Exception e) {
e.printStackTrace();
}finally {
try {
/*释放*/
System.out.println(Thread.currentThread().getName() + "释放了锁");
mutex.release();
} catch (Exception e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
/*定义信号灯。只能允许10个线程并发操作*/
final Semaphore semaphore = new Semaphore(11);
for (int i = 1; i <= 20; i++) {
Runnable runnable =new Runnable() {
@Override
public void run() {
try {
semaphore.acquire();
/*链接zk*/
CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(CONNECT_SERVER, SESSION_TIMEOUT, CONNECTION_TIMEOUT, new RetryNTimes(10, 5000));
curatorFramework.start();
doLock(curatorFramework);
semaphore.release();
} catch (Exception e) {
e.printStackTrace();
}
}
};
executorService.execute(runnable);
}
}
}
