オンライン実戦を経たredis分布式ロックとzookeeper分布式ロックの違い
オンライン実戦を経たredis分散ロックコード.
使えますが、性能が悪いです.
検討:
1.ロックを持つスレッドのみでロックを解除できます
2. ノードとタイムアウト時間を同じkeyで設定
考慮されていません:
1. 再入力不可
2. ローカルロックがないと、コンカレント競合が多いシーンでは使用されず、コンカレントパフォーマンスが低下します.ローカルロック非スピン
3. ロック待ちソートは考慮されていません.これはredisでは実現しにくい.
redisのlistで実現できるが、listでは各サブノードにタイムアウト時間がないという欠点がある.redisもファジイクエリkey*を行うことができない.
だからzookeeperで実現したほうがいい.しかしzookeeperは性能のボトルネックに直面して、私達のオフラインの出現、いつも登録できない情況.
zookeeperの原理は一時ノードです
使用方法:
GlobalLockRedisImpl globalLockRedis = new GlobalLockRedisImpl(casRedis, maxLockSeconds, sleepTimeMillis, redisKey);//分散ロックglobalLockRedisを取得する.lock();
try{
//do something
}finnaly{
//分散ロックglobalLockRedis.unlock(); }
private boolean setIfAbsent(String key, String value, int expireMilliSeconds) { String result = casRedis.set(key, value, "nx", "px", expireMilliSeconds); if (result != null && result.equalsIgnoreCase("OK")) { return true; } return false;
コード:
バージョン2、オンラインで検証されていません.
使えますが、性能が悪いです.
検討:
1.ロックを持つスレッドのみでロックを解除できます
2. ノードとタイムアウト時間を同じkeyで設定
考慮されていません:
1. 再入力不可
2. ローカルロックがないと、コンカレント競合が多いシーンでは使用されず、コンカレントパフォーマンスが低下します.ローカルロック非スピン
3. ロック待ちソートは考慮されていません.これはredisでは実現しにくい.
redisのlistで実現できるが、listでは各サブノードにタイムアウト時間がないという欠点がある.redisもファジイクエリkey*を行うことができない.
だからzookeeperで実現したほうがいい.しかしzookeeperは性能のボトルネックに直面して、私達のオフラインの出現、いつも登録できない情況.
zookeeperの原理は一時ノードです
使用方法:
GlobalLockRedisImpl globalLockRedis = new GlobalLockRedisImpl(casRedis, maxLockSeconds, sleepTimeMillis, redisKey);//分散ロックglobalLockRedisを取得する.lock();
try{
//do something
}finnaly{
//分散ロックglobalLockRedis.unlock(); }
private boolean setIfAbsent(String key, String value, int expireMilliSeconds) { String result = casRedis.set(key, value, "nx", "px", expireMilliSeconds); if (result != null && result.equalsIgnoreCase("OK")) { return true; } return false;
コード:
/**
* redis , .
*
* @author loufei
*
* 2015-5-28
*/
public class GlobalLockRedisImpl implements GlobalLock {
private static ILog LOGGER = LogFactory.getLog(GlobalLockRedisImpl.class);
private final CasRedisService casRedis;
private final int maxLockSeconds;
private final long sleepTimeMillis;
private Thread exclusiveOwnerThread;
private final String key;
public GlobalLockRedisImpl(CasRedisService casRedis, int maxLockSeconds, long sleepTimeMillis, String key) {
this.casRedis = casRedis;
this.maxLockSeconds = maxLockSeconds;
if (maxLockSeconds > 30) {
maxLockSeconds = 30;
}
this.sleepTimeMillis = sleepTimeMillis;
if (sleepTimeMillis > 1000) {
sleepTimeMillis = 1000;
}
this.key = key;
}
@Override
public void lock() {
long startTime = System.currentTimeMillis();
int tryCount = 0;
while (true) {
tryCount++;
// setIfAbsent tps .
Boolean result = casRedis.setIfAbsent(key, "1", maxLockSeconds * 1000);
// , , timeOut. , key , .
if (result == null || !result) {// ,
LOGGER.info(" spin ,key=" + key + ",tryCount=" + tryCount);
try {
Thread.sleep(sleepTimeMillis);
} catch (InterruptedException e) {
// nothing need to be done
}
long end = System.currentTimeMillis();
//
long costTime = end - startTime;
// ( 3 ), key .
long sleepMillisWaterMark = TimeUnit.SECONDS.toMillis(2);
if (costTime > sleepMillisWaterMark) {
String message = "get redis global lock error .1. compete failed 2. key do not set timeOut ,exist for ever ,maxLockSeconds="
+ maxLockSeconds
+ ",costTimeMillis="
+ costTime
+ ",key="
+ key
+ ",retryCount="
+ tryCount
+ ",sleepTimeMillis="
+ sleepTimeMillis
+ ",costTimePerTime="
+ (costTime / ((double) tryCount));
// , .
throw new GlobalLockTimeOutException(message);
}
continue;
}
exclusiveOwnerThread = Thread.currentThread();
break;
}
long endTime = System.currentTimeMillis();
long costTime = endTime - startTime;
LOGGER.info("get redis global lock success,maxLockSeconds=" + maxLockSeconds + ",costTimeMillis=" + costTime
+ ",key=" + key + ",retryCount=" + tryCount + ",sleepTimeMillis=" + sleepTimeMillis
+ ",costTimePerTime=" + costTime / ((double) tryCount) + ",startTime=" + startTime + ",endTime="
+ endTime);
}
@Override
public void unlock() {
long startTime = System.currentTimeMillis();
if (exclusiveOwnerThread == Thread.currentThread()) {
Integer del = casRedis.del(key.getBytes());
long endTime = System.currentTimeMillis();
exclusiveOwnerThread = null;
LOGGER.info(" global unlock,del count=" + del + ".key=" + key + " ,costTime millis="
+ (endTime - startTime) + ",startTime=" + startTime + ",endTime=" + endTime);
} else {
LOGGER.info(" thread do not get lock ,can not unlock. key=" + key + ",exclusiveOwnerThread="
+ exclusiveOwnerThread + ",current thrad=" + Thread.currentThread());
}
}
public int getMaxLockSeconds() {
return maxLockSeconds;
}
}
private boolean setIfAbsent(String key, String value, int expireMilliSeconds) {
String result = casRedis.set(key, value, "nx", "px", expireMilliSeconds);
if (result != null && result.equalsIgnoreCase("OK")) {
return true;
}
return false;
}バージョン2、オンラインで検証されていません.
/**
* redis , .
*
* @author loufei
*
* 2015-5-28
*/
public class GlobalLockSeqRedisImpl implements GlobalLock {
private static long maxWaitSeconds = 2;
private static final String split = "____";
private static AtomicInteger seqCount = new AtomicInteger();
private static String host = null;
private static ILog LOGGER = LogFactory.getLog(GlobalLockRedisImpl.class);
private final static ExecutorService threadPoolExecutor = TaxiExecutors
.newCachedThreadPool(new ThreadFactoryBuilder("GlobalLockRedisImpl"));
private final didikuaidi.redis.clients.jedis.JedisCommands casRedis;
private final int maxLockSeconds;
private final long sleepTimeMillis;
private Thread exclusiveOwnerThread;
private final String key;
private final Lock lock = new ReentrantLock();
private String value = null;
static {
InetAddress localHost = null;
try {
localHost = InetAddress.getLocalHost();
} catch (UnknownHostException e) {
throw new RuntimeException(e);
}
host = (UUID.randomUUID() + localHost.getHostAddress()).replace(split, "");
}
public GlobalLockSeqRedisImpl(JedisCommands casRedis, int maxLockSeconds, long sleepTimeMillis, String key) {
this.casRedis = casRedis;
this.maxLockSeconds = maxLockSeconds;
if (maxLockSeconds > 30) {
maxLockSeconds = 30;
}
this.sleepTimeMillis = sleepTimeMillis;
if (sleepTimeMillis > 1000) {
sleepTimeMillis = 1000;
}
this.key = key;
}
private boolean setIfAbsent(String key, String value, int expireMilliSeconds) {
String result = casRedis.set(key, value, "nx", "px", expireMilliSeconds);
if (result != null && result.equalsIgnoreCase("OK")) {
return true;
}
return false;
}
private int lockRedisSeq(long startMillis) {
int seqNo = seqCount.incrementAndGet();
value = host + split + new Date().getTime() + split + seqNo;
Long listSize = casRedis.lpush(key, value);
int count = 1;
LOGGER.info("listSize=" + listSize);
if (listSize.longValue() == 1l) {
// 1, .
return count;
} else {
// , error
if (listSize > DynamicConfig.getInt(DynamicConfigKeys.KUAIPAY_LOCK_LIMIT, 21)) {
LOGGER.error("too_much_lock_node listSize=" + listSize);
} else {
LOGGER.info("lock_queue_size=" + listSize + ",key=" + key);
}
//
while (true) {
try {
Thread.sleep(sleepTimeMillis);
} catch (InterruptedException e) {
e.printStackTrace();
}
// -1 .FIFO
String tailNodeValue = casRedis.lindex(key, -1);
if (tailNodeValue == null) {
//
String message = "tailNodeValue is null ,all node is delete inclued itself key=" + key;
LOGGER.error(message);
throw new GlobalLockTimeOutException(message);
} else if (tailNodeValue.equals(value)) {
//
return count;
} else {
String[] split = tailNodeValue.split(GlobalLockSeqRedisImpl.split);
String host = split[0];
//
Date nodeCreateTime = new Date(Long.valueOf(split[1]));
long diffInMillis = new Date().getTime() - nodeCreateTime.getTime();
long maxLockTime = TimeUnit.MINUTES.toMillis(1);
if (host.equals(this.host) && diffInMillis > TimeUnit.SECONDS.toMillis(3)) {
// ( ), 3 , .
lrem(tailNodeValue);
}
if (diffInMillis > TimeUnit.SECONDS.toMillis(30) && diffInMillis <= maxLockTime) {
// error , . ip . java
LOGGER.error("Lock Node donot unLock , ,key=" + key + " unlockNodeIp= " + split[0]
+ ",nodeCreateTime=" + nodeCreateTime);
} else if (diffInMillis > maxLockTime) {
//
LOGGER.error("Lock Node donot unLock , ,key=" + key + " unlockNodeIp= " + split[0]
+ ",nodeCreateTime=" + nodeCreateTime);
lrem(tailNodeValue);
}
}
long now = System.currentTimeMillis();
long costTime = now - startMillis;
count++;
if (costTime > TimeUnit.SECONDS.toMillis(maxLockSeconds)) {
String errorString = getErrorString(count, costTime);
throw new GlobalLockTimeOutException(errorString);
}
}
}
//
}
private void lrem(String tailNodeValue) {
// -1 .FIFO . rem ,
long remCount = casRedis.lrem(key, -2, tailNodeValue);
if (remCount != 1) {
LOGGER.error("del count dot not euqal 1,remCount=" + remCount);
}
}
@Override
public void lock() {
long startTime = System.currentTimeMillis();
// ,
try {
boolean success = lock.tryLock(maxWaitSeconds, TimeUnit.SECONDS);
if (!success) {
long end = System.currentTimeMillis();
//
long costTime = end - startTime;
String message = getErrorString(1, costTime);
LOGGER.error("lock timeOut " + message);
throw new GlobalLockTimeOutException(message);
}
} catch (Exception e) {
long end = System.currentTimeMillis();
//
long costTime = end - startTime;
String message = getErrorString(1, costTime);
LOGGER.error("lock meet exception " + message, e);
throw new GlobalLockTimeOutException(message);
}
int tryCount = lockRedisSeq(startTime);
exclusiveOwnerThread = Thread.currentThread();
long endTime = System.currentTimeMillis();
long costTime = endTime - startTime;
LOGGER.info("get redis global lock success,maxLockSeconds=" + maxLockSeconds + ",costTimeMillis=" + costTime
+ ",key=" + key + ",retryCount=" + tryCount + ",sleepTimeMillis=" + sleepTimeMillis
+ ",costTimePerTime=" + costTime / ((double) tryCount) + ",startTime=" + startTime + ",endTime="
+ endTime);
}
private int lockRedis(long startTime) {
// , ,
int tryCount = 0;
while (true) {
tryCount++;
// setIfAbsent tps .
Boolean result = this.setIfAbsent(key, "1", maxLockSeconds * 1000);
// , , timeOut. , key , .
if (result == null || !result) {// ,
LOGGER.info(" spin ,key=" + key + ",tryCount=" + tryCount);
try {
Thread.sleep(sleepTimeMillis);
} catch (InterruptedException e) {
// nothing need to be done
}
long end = System.currentTimeMillis();
//
long costTime = end - startTime;
// ( 3 ), key .
long sleepMillisWaterMark = TimeUnit.SECONDS.toMillis(maxWaitSeconds);
if (costTime > sleepMillisWaterMark) {
String message = getErrorString(tryCount, costTime);
// , .
throw new GlobalLockTimeOutException(message);
}
continue;
}
exclusiveOwnerThread = Thread.currentThread();
break;
}
return tryCount;
}
private String getErrorString(int tryCount, long costTime) {
return "get redis global lock error .1. compete failed 2. key do not set timeOut ,exist for ever ,maxLockSeconds="
+ maxLockSeconds + ",costTimeMillis=" + costTime + ",key=" + key + ",retryCount=" + tryCount
+ ",sleepTimeMillis=" + sleepTimeMillis + ",costTimePerTime=" + (costTime / ((double) tryCount));
}
@Override
public void unlock() {
long startTime = System.currentTimeMillis();
try {
lock.unlock();
} catch (IllegalMonitorStateException e) {
LOGGER.error("IllegalMonitorStateException", e);
}
if (exclusiveOwnerThread == Thread.currentThread()) {
// , ,
Long delCount = unlockSeqRedisAndRetryIfError();
long endTime = System.currentTimeMillis();
exclusiveOwnerThread = null;
LOGGER.info(" global unlock,del count=" + delCount + ".key=" + key + " ,costTime millis="
+ (endTime - startTime) + ",startTime=" + startTime + ",endTime=" + endTime);
} else {
LOGGER.info(" thread do not get lock ,can not unlock. key=" + key + ",exclusiveOwnerThread="
+ exclusiveOwnerThread + ",current thrad=" + Thread.currentThread());
}
}
private Long unlockSeqRedisAndRetryIfError() {
Long delCount = 0l;
try {
delCount = unlockSeqRedisLock();
} catch (Exception e) {
LOGGER.error("unlockSeqRedisLock error", e);
threadPoolExecutor.execute(getRetryUnlockTask());
}
return delCount;
}
private Runnable getRetryUnlockTask() {
return new Runnable() {
@Override
public void run() {
sleepSlience();
for (int i = 0; i < 3; i++) {
try {
unlockSeqRedisLock();
} catch (Exception e) {
// ,
LOGGER.error("unlockSeqRedisLock error", e);
sleepSlience();
continue;
}
break;
}
}
private void sleepSlience() {
try {
Thread.sleep(1000l);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
}
};
}
@Deprecated
// unlockSeqRedisAndRetryIfError
private Long unlockSeqRedisLock() {
// -1 .FIFO
String lastNodeValue = casRedis.rpop(key);
if (!value.equals(lastNodeValue)) {
LOGGER.error("del result do not match expect value,expect=" + value + ",acutal value=" + lastNodeValue);
// Rpush, , .
casRedis.lpush(key, lastNodeValue);
Long lrem = casRedis.lrem(key, -2, value);
if (lrem != 1) {
LOGGER.error("del resutl do not match expect value,expect" + value);
}
return lrem;
}
return 1l;
}
private Long unlockRedisLock() {
return casRedis.del(key);
}
public int getMaxLockSeconds() {
return maxLockSeconds;
}
}