分布式鎖用Redis好？還是Zookeeper好？
Redis 實現(xiàn)
//基于jedis和lua腳本來實現(xiàn)
privatestaticfinal String LOCK_SUCCESS = "OK";
privatestaticfinal Long RELEASE_SUCCESS = 1L;
privatestaticfinal String SET_IF_NOT_EXIST = "NX";
privatestaticfinal String SET_WITH_EXPIRE_TIME = "PX";
@Override
public String acquire() {
try {
// 獲取鎖的超時時間，超過這個時間則放棄獲取鎖
long end = System.currentTimeMillis() + acquireTimeout;
// 隨機生成一個 value
String requireToken = UUID.randomUUID().toString();
while (System.currentTimeMillis() < end) {
String result = jedis
.set(lockKey, requireToken, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
if (LOCK_SUCCESS.equals(result)) {
return requireToken;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
} catch (Exception e) {
log.error("acquire lock due to error", e);
}
returnnull;
}
@Override
public boolean release(String identify) {
if (identify == null) {
returnfalse;
}
//通過lua腳本進行比對刪除操作，保證原子性
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = new Object();
try {
result = jedis.eval(script, Collections.singletonList(lockKey),
Collections.singletonList(identify));
if (RELEASE_SUCCESS.equals(result)) {
log.info("release lock success, requestToken:{}", identify);
returntrue;
}
} catch (Exception e) {
log.error("release lock due to error", e);
} finally {
if (jedis != null) {
jedis.close();
}
}
log.info("release lock failed, requestToken:{}, result:{}", identify, result);
returnfalse;
}
Redisson 實現(xiàn)
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.4</version>
</dependency>
private void test() {
//分布式鎖名 鎖的粒度越細，性能越好
RLock lock = redissonClient.getLock("test_lock");
lock.lock();
try {
//具體業(yè)務......
} finally {
lock.unlock();
}
}
// 最常見的使用方法
lock.lock();
// 加鎖以后10秒鐘自動解鎖
// 無需調(diào)用unlock方法手動解鎖
lock.lock(10, TimeUnit.SECONDS);
繼續(xù)點進 scheduleExpirationRenewal 方法：
點進 renewExpiration 方法：
小結：雖然 lock() 有自動續(xù)鎖機制，但是開發(fā)中還是推薦使用 lock(time，timeUnit)，因為它省掉了整個續(xù)期帶來的性能損，可以設置過期時間長一點，搭配 unlock()。
若業(yè)務執(zhí)行完成，會手動釋放鎖，若是業(yè)務執(zhí)行超時，那一般我們服務也都會設置業(yè)務超時時間，就直接報錯了，報錯后就會通過設置的過期時間來釋放鎖。
public void test() {
RLock lock = redissonClient.getLock("test_lock");
lock.lock(30, TimeUnit.SECONDS);
try {
//.......具體業(yè)務
} finally {
//手動釋放鎖
lock.unlock();
}
}
基于 Zookeeper 來實現(xiàn)分布式鎖
當?shù)谝粋€線程進來時會去父節(jié)點上創(chuàng)建一個臨時的順序節(jié)點。
第二個線程進來發(fā)現(xiàn)鎖已經(jīng)被持有了，就會為當前持有鎖的節(jié)點注冊一個 watcher 監(jiān)聽器。
第三個線程進來發(fā)現(xiàn)鎖已經(jīng)被持有了，因為是順序節(jié)點的緣故，就會為上一個節(jié)點去創(chuàng)建一個 watcher 監(jiān)聽器。
當?shù)谝粋€線程釋放鎖后，刪除節(jié)點，由它的下一個節(jié)點去占有鎖。
public class ZooKeeperDistributedLock implements Watcher {
private ZooKeeper zk;
private String locksRoot = "/locks";
private String productId;
private String waitNode;
private String lockNode;
private CountDownLatch latch;
private CountDownLatch connectedLatch = new CountDownLatch(1);
private int sessionTimeout = 30000;
public ZooKeeperDistributedLock(String productId) {
this.productId = productId;
try {
String address = "192.168.189.131:2181,192.168.189.132:2181";
zk = new ZooKeeper(address, sessionTimeout, this);
connectedLatch.await();
} catch (IOException e) {
throw new LockException(e);
} catch (KeeperException e) {
throw new LockException(e);
} catch (InterruptedException e) {
throw new LockException(e);
}
}
public void process(WatchedEvent event) {
if (event.getState() == KeeperState.SyncConnected) {
connectedLatch.countDown();
return;
}
if (this.latch != null) {
this.latch.countDown();
}
}
public void acquireDistributedLock() {
try {
if (this.tryLock()) {
return;
} else {
waitForLock(waitNode, sessionTimeout);
}
} catch (KeeperException e) {
throw new LockException(e);
} catch (InterruptedException e) {
throw new LockException(e);
}
}
//獲取鎖
public boolean tryLock() {
try {
// 傳入進去的locksRoot + “/” + productId
// 假設productId代表了一個商品id，比如說1
// locksRoot = locks
// /locks/10000000000，/locks/10000000001，/locks/10000000002
lockNode = zk.create(locksRoot + "/" + productId, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
// 看看剛創(chuàng)建的節(jié)點是不是最小的節(jié)點
// locks：10000000000，10000000001，10000000002
List<String> locks = zk.getChildren(locksRoot, false);
Collections.sort(locks);
if(lockNode.equals(locksRoot+"/"+ locks.get(0))){
//如果是最小的節(jié)點,則表示取得鎖
return true;
}
//如果不是最小的節(jié)點，找到比自己小1的節(jié)點
int previousLockIndex = -1;
for(int i = 0; i < locks.size(); i++) {
if(lockNode.equals(locksRoot + “/” + locks.get(i))) {
previousLockIndex = i - 1;
break;
}
}
this.waitNode = locks.get(previousLockIndex);
} catch (KeeperException e) {
throw new LockException(e);
} catch (InterruptedException e) {
throw new LockException(e);
}
return false;
}
private boolean waitForLock(String waitNode, long waitTime) throws InterruptedException, KeeperException {
Stat stat = zk.exists(locksRoot + "/" + waitNode, true);
if (stat != null) {
this.latch = new CountDownLatch(1);
this.latch.await(waitTime, TimeUnit.MILLISECONDS);
this.latch = null;
}
return true;
}
//釋放鎖
public void unlock() {
try {
System.out.println("unlock " + lockNode);
zk.delete(lockNode, -1);
lockNode = null;
zk.close();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
//異常
public class LockException extends RuntimeException {
private static final long serialVersionUID = 1L;
public LockException(String e) {
super(e);
}
public LockException(Exception e) {
super(e);
}
}
}
既然明白了 Redis 和 ZK 分別對分布式鎖的實現(xiàn)，那么總該有所不同的吧。沒錯，我都幫大家整理好了：
實現(xiàn)方式的不同，Redis 實現(xiàn)為去插入一條占位數(shù)據(jù)，而 ZK 實現(xiàn)為去注冊一個臨時節(jié)點。
遇到宕機情況時，Redis 需要等到過期時間到了后自動釋放鎖，而 ZK 因為是臨時節(jié)點，在宕機時候已經(jīng)是刪除了節(jié)點去釋放鎖。
Redis 在沒搶占到鎖的情況下一般會去自旋獲取鎖，比較浪費性能，而 ZK 是通過注冊監(jiān)聽器的方式獲取鎖，性能而言優(yōu)于 Redis。
最近給大家找了 JVM學習視頻
資源，怎么領?。?/span>
掃二維碼為，加我微信，回復：JVM
注意，不要亂回復 沒錯，不是機器人 記得一定要等待，等待才有好東西
評論
圖片
表情