zk使用的場景：分佈式協調、分佈式鎖、服務註冊node
分佈式協調，A註冊監聽到某節點，B系統修改該節點的數據，能夠進行協調redis
分佈式鎖，獲取不到鎖，註冊個監聽器便可，不須要不斷主動嘗試獲取鎖，性能開銷較小，另一點就是，若是是redis獲取鎖的那個客戶端bug了或者掛了，那麼只能等待超時時間以後才能釋放鎖；而zk的話，由於建立的是臨時znode，只要客戶端掛了，znodspring
public class ZooKeeperDistributedLock implements Watcher{緩存
private ZooKeeper zk;服務器
private String locksRoot= "/locks";session
private String productId;app
private String waitNode;分佈式
private String lockNode;性能
private CountDownLatch latch;ui
private CountDownLatch connectedLatch = new CountDownLatch(1);
private int sessionTimeout = 30000;
public ZooKeeperDistributedLock(String productId){
this.productId = productId;
try {
String address = "192.168.31.187:2181,192.168.31.19:2181,192.168.31.227: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);
// 看看剛建立的節點是否是最小的節點
// locks：10000000000，10000000001，10000000002
List<String> locks = zk.getChildren(locksRoot, false);
Collections.sort(locks);
if(lockNode.equals(locksRoot+"/"+ locks.get(0))){
//若是是最小的節點,則表示取得鎖
return true;
}
//若是不是最小的節點，找到比本身小1的節點
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 {
// 刪除/locks/10000000000節點
// 刪除/locks/10000000001節點
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);
}
}
// 若是有一把鎖，被多我的給競爭，此時多我的會排隊，第一個拿到鎖的人會執行，而後釋放鎖，後面的每一個人都會去監聽排在本身前面的那我的建立的node上，一旦某我的釋放了鎖，排在本身後面的人就會被zookeeper給通知，一旦被通知了以後，就ok了，本身就獲取到了鎖，就能夠執行代碼了
========================================================================================
PERSISTENT(0, false, false),持久節點 PERSISTENT_SEQUENTIAL(2, false, true),持久有序節點 EPHEMERAL(1, true, false),臨時節點 EPHEMERAL_SEQUENTIAL(3, true, true);臨時有序節點
//節點路徑、節點數據、權限、節點類型，當前爲持久節點
String create = zooKeeper.create("/kxj","kxj".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//獲取子節點
List<String> children = zooKeeper.getChildren("/", false);
//判斷節點是否存在
Stat exists = zooKeeper.exists("/kxj", true);
//獲取子節點
zk.getChildren("/create",false);
//修改是數據
zk.setData("/create","nihao".getBytes(),-1);//最後一個是版本號，CAS原來，若是不使用版本號使用-1
===========================================================================================
分佈式系統中共享session，使用spring session+redis
//準備工做
啓用redis管理Session
//第一種方式：修改application.properties，添加
spring.session.store-type=redis
//第二種方式：啓用註解
@@EnableRedisHttpSession(maxInactiveIntervalInSeconds=2000) 修改過時時間單位是秒
//設置緩存
public Object setSession(HttpServletRequest request,String sessionKey,String sessionValue){
HttpSession httpSession=request.getSession();
httpSession.setAttribute(sessionKey,sessionValue);//保存session到redis中
return "success,port:" + Port;
}
//獲取緩存
@RequestMapping("/getSession")
public Object getSession(HttpServletRequest request,String sessionKey){
String key = httpSession.getAttribute(sessionKey);//獲取redis中的session
return "sessionValue:" + key;
}
pom文件
<dependency>
<groupId>org.springframework.session</groupId>
<artifactId>spring-session-data-redis</artifactId>
</dependency>
配置文件中
spring.redis.host=192.168.118.3 spring.redis.port=6379 #服務器端口 server.port=8888