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