1.zookeeper安装

curl -sSL https://get.daocloud.io/daotools/set_mirror.sh | sh -s http://193f7471.m.daocloud.iodaocloud123zkCli.sh -server 127.0.0.1:3001 create /minitest hellominiget /adminls /admindelete /admin

dataVersion 可以作为乐观锁

1.1 zookeeper基础知识

(1) session基本原理

​ 客户端与服务器之间维持会话，每个会话都可以设置一个超时时间，心跳结束那么会话就结束。心跳存在,那么就是告诉服务器别把我删掉。

​ session过期之后，session创建的临时节点会被抛弃。

心跳机制：客户端向服务器发送Ping包请求。

(2)基本命令

create /aaa bbb 创建一个aaa节点 值为bbb

get /aaa 获取节点

set /aaa bbb 1 设置aaa节点值为bbb，并且只有在版本为1的情况下生效，否则失败

delete /aaa 注意若节点下没有子节点才可以删除。若存在子节点，那么先删除子节点。

(3)zk特性watcher机制

针对每个节点的操作，都会有一个监听者watcher。

当监控的某个对象发生了变化(父节点或子节点增删改操作)，就会出发watcher机制。

针对不同类型的操作，触发的watcher事件也不同。（创建，删除，数据变化事件）

zk中的watcher是一次性的，触发后就会被销毁。

stat /aaa watch 监听aaa

create /aaa 123 创建一个aaa节点

watcherEvent : node created path:/aaa

get /aaa watcher 设置 aaa监听

set /aaa bbb

触发watcherEvent nodeDataChanged path /aaa

delete /imooc 发现没有watcher事件，则说明watcher是一次性的

get /aaa watcher 设置 aaa监听 delete /imooc

触发watcherEvent NodeDeleted

ls为父节点设置watcher，创建子节点触发NodeChildrenChanged

ls为父节点设置watcher, 删除子节点触发NodeChildrenChanged

ls为父节点设置watcher, 修改子节点值不触发事件，只有把子节点当父节点，修改数据得时候才发生nodeChanged事件

2. 分布式锁依赖监听：

(1)将zookeeper添加到springboot中

@Component("applicationContextHelper")
public class ApplicationContextHelper implements ApplicationContextAware {

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        SpringHelper.setApplicationContext(applicationContext);
        InitData.start();
        new Thread(new RebuildCacheThread()).start();
        ZooKeeperSession.init();
    }
}

(2）分布式锁

public class ZooKeeperSession {
	
	private static CountDownLatch connectedSemaphore = new CountDownLatch(1);
	
	private ZooKeeper zookeeper;

	public ZooKeeperSession() {
		// 去连接zookeeper server，创建会话的时候，是异步去进行的
		// 所以要给一个监听器，说告诉我们什么时候才是真正完成了跟zk server的连接
		try {
			this.zookeeper = new ZooKeeper(
					"127.0.0.1:3001,127.0.0.1:3002,127.0.0.1:3003", 
					50000, 
					new ZooKeeperWatcher());
			// 给一个状态CONNECTING，连接中
			System.out.println(zookeeper.getState());
			
			try {
				// CountDownLatch
				// java多线程并发同步的一个工具类
				// 会传递进去一些数字，比如说1,2 ，3 都可以
				// 然后await()，如果数字不是0，那么久卡住，等待
				
				// 其他的线程可以调用coutnDown()，减1
				// 如果数字减到0，那么之前所有在await的线程，都会逃出阻塞的状态
				// 继续向下运行
				
				connectedSemaphore.await();
			} catch(InterruptedException e) {
				e.printStackTrace();
			}

			System.out.println("ZooKeeper session established......");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 获取分布式锁
	 * @param productId
	 */
	public void acquireDistributedLock(Long productId) {
		String path = "/product-lock-" + productId;
	
		try {
			zookeeper.create(path, "".getBytes(), 
					Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
			System.out.println("success to acquire lock for product[id=" + productId + "]");  
		} catch (Exception e) {
			// 如果那个商品对应的锁的node，已经存在了，就是已经被别人加锁了，那么就这里就会报错
			// NodeExistsException
			int count = 0;
			while(true) {
				try {
					Thread.sleep(1000); 
					zookeeper.create(path, "".getBytes(), 
							Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
				} catch (Exception e2) {
					count++;
					System.out.println("the " + count + " times try to acquire lock for product[id=" + productId + "]......");
					continue;
				}
				System.out.println("success to acquire lock for product[id=" + productId + "] after " + count + " times try......");
				break;
			}
		}
	}
	
	/**
	 * 释放掉一个分布式锁
	 * @param productId
	 */
	public void releaseDistributedLock(Long productId) {
		String path = "/product-lock-" + productId;
		try {
			zookeeper.delete(path, -1); 
			System.out.println("release the lock for product[id=" + productId + "]......");  
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 建立zk session的watcher
	 * @author Administrator
	 *
	 */
	private class ZooKeeperWatcher implements Watcher {

		public void process(WatchedEvent event) {
			System.out.println("Receive watched event: " + event.getState());
			if(KeeperState.SyncConnected == event.getState()) {
				connectedSemaphore.countDown();
			} 
		}
		
	}
	
	/**
	 * 封装单例的静态内部类
	 * @author Administrator
	 *
	 */
	private static class Singleton {
		
		private static ZooKeeperSession instance;
		
		static {
			instance = new ZooKeeperSession();
		}
		
		public static ZooKeeperSession getInstance() {
			return instance;
		}
		
	}
	
	/**
	 * 获取单例
	 * @return
	 */
	public static ZooKeeperSession getInstance() {
		return Singleton.getInstance();
	}
	
	/**
	 * 初始化单例的便捷方法
	 */
	public static void init() {
		getInstance();
	}
	
}

（3）以分布式环境下修改缓存为例(这里代码用到了2级缓存：redis+ehcache)

public class RebuildCacheThread implements Runnable {
	
	public void run() {
		RebuildCacheQueue rebuildCacheQueue = RebuildCacheQueue.getInstance();
		ZooKeeperSession zkSession = ZooKeeperSession.getInstance();
		IProductService cacheService = (IProductService) SpringHelper.popBean(IProductService.class);
		
		while(true) {
			Product productInfo = rebuildCacheQueue.takeProductInfo();
			
			zkSession.acquireDistributedLock(productInfo.getId());  
			
			Product existedProductInfo = cacheService.getProductByRedisCache(productInfo.getId());
			
			if(existedProductInfo != null) {
				// 比较当前数据的时间版本比已有数据的时间版本是新还是旧
				try {
					if(productInfo.getUpdateTime().before(existedProductInfo.getUpdateTime())) {
						System.out.println("current date[" + productInfo.getUpdateTime() + "] is before existed date[" + existedProductInfo.getUpdateTime() + "]");
						continue;
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
				System.out.println("current date[" + productInfo.getUpdateTime() + "] is after existed date[" + existedProductInfo.getUpdateTime() + "]");
			} else {
				System.out.println("existed product info is null......");   
			}
			cacheService.saveProduct2RedisCache(productInfo);  
			zkSession.releaseDistributedLock(productInfo.getId()); 
		}
	}

}