- 👏作者简介：大家好，我是爱吃芝士的土豆倪，24届校招生Java选手，很高兴认识大家
- 📕系列专栏：Spring源码、JUC源码、Kafka原理、分布式技术原理
- 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
- 🍂博主正在努力完成2023计划中：源码溯源，一探究竟
- 📝联系方式：nhs19990716，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀
文章目录
- Zookeeper watcher
- 为什么学习源码
- watcher
- 图解
- 源码实现
Zookeeper watcher
为什么学习源码
公司要调用一个第三方的接口进行一个操作，但是这个操作可能会比较耗时，需要一段时间来响应，在第三方接口不能去修改的情况下，想去提升性能。
在这个过程中我们有很多方法去解决它，包括定时轮询，做事件响应机制，异步化的方式，即便第三方没办法修改的情况下，我们也可以在自己这边形成一个循环，形成一个高效的循环去处理这个调用。
在分析问题的时候，我们所学的事件的机制、异步化、基于线程的生产者消费者模型，其实它可以在刚刚的场景中去使用，比如我们发起这个请求，他可以将其放置在队列中，通过线程去和对应的接口做远程通信，这时我们可以解决我们这端的并发量的问题，他们第三方在不改变的情况下，我们这端某种程度上提升了一定的吞吐量，然后基于事件的机制就是我们调用第三方接口给返回的时候，基于这个返回结果发送一个通知去告诉调用者结果处理完了，你去拿这个结果做后续的处理
watcher
在api中，如果我们的客户端需要去实现watcher，就想zk做注册中心，配置中心的情况下，我们都需要实现在zk server上的配置变更和服务地址变更的通知 要去告诉我们的客户端，所有的客户端，你的数据发生了变化你需要采取一些行动，其实这就是一个通知的机制。
// standrd 标准监听 (一次性监听)
ZooKeeper zooKeeper=new ZooKeeper("192.168.216.128:2181", 5000, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
//表示连接成功之后，会产生的回调时间
}
});
Stat stat=new Stat();
zooKeeper.getData("/first", new DataWatchListener(),stat); //针对当前节点
class DataWatchListener implements Watcher{
@Override
public void process(WatchedEvent watchedEvent) {
String path=watchedEvent.getPath();
//再次注册监听
try {
zooKeeper.getData(path,this,new Stat());
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
// 在3.6.1 还有着新的监听 持久化监听 和 持久化递归监听
// 持久化监听：只需要注册一次事件
// 持久化递归监听：其子节点发生变化，都会触发监听
// 默认情况下，是递归持久化监听
ZooKeeper.addWatch("path",new DataWatchListener(),Add.WatchMode.PERSISTENT_RECURSIVE)
图解
首先客户端要发起一个请求，客户端所有的请求先发到阻塞队列中，然后一个 SendThread线程 去轮询队列，通过take的方式，可以发现异步的方式能够很大程度上的提升整个的处理性能，发送过来的任务是一个request，它里面可以包括 crud exist等，我们后续的请求实际上就是NIO，实际上会将请求转换成序列化以后，自己会实现一个序列化机制，然后将这个请求发送到服务端，这个请求要做的就是 注册 watcher 带的内容是 path/watch:true,发送到服务端之后，因为客户端和服务端建立好连接以后，会维持这个会话，所以最终服务端会保存这个watcher，将其存储在HashMap中。
之所以是HashMap<String,HashSet> 是因为可能有多个客户端。
当注册成功会给一个返回，并且这个动作是异步化的操作。
然后客户端这边也需要存储一个事件的管理 提供了一个类 ZkWatcherManager，通过一些集合去存储客户端那边锁对应的watcher
当我们通过xshell来修改zk上对应监听的路径节点时，此时会触发对应节点的变化，服务端会判断这个节点是否存在监听，也就是在这个hashmap中查询是否有watcher。
因为之间说过zk服务端和客户端实际上存在着一个会话，所以，最后直接push发送给客户端即可
通过上图可以发现，所有和网络通信相关的，都是采用异步（生产者消费者模型）
当然上图中还有一些地方没有说清除，具体如下：
当ZooKeeper服务器检测到某个Znode的状态发生变化时，会向对该Znode注册了Watcher的客户端发送一个通知消息。这个通知消息包含了变化的类型（例如数据内容的变化、子节点的变化等）以及该Znode的最新状态信息。
当客户端接收到Watcher通知后，它会根据通知消息中的信息进行相应的处理。具体地，如果客户端的ZkWatcherManager中有对应的Watcher对象，就会调用该Watcher的process方法来处理通知消息。在process方法中，客户端可以根据变化的类型和最新的状态信息来进行相应的操作。
举个例子，假设客户端在启动时对一个Znode注册了一个数据内容变化的Watcher。当该Znode的数据内容发生变化时，ZooKeeper服务器会向该客户端发送一个通知消息，并在消息中指明变化的类型是“数据内容的变化”，同时包含最新的数据内容。客户端的ZkWatcherManager会找到对应的Watcher对象，并调用它的process方法。在process方法中，客户端可以根据最新的数据内容进行相应的操作，比如重新获取数据内容、重新注册Watcher等。
源码实现
zookeeper.exists();
---------------------------------------------------------------------------------------
public Stat exists(final String path, Watcher watcher) throws KeeperException, InterruptedException {
final String clientPath = path;
PathUtils.validatePath(clientPath); //节点校验
// the watch contains the un-chroot path
WatchRegistration wcb = null;
if (watcher != null) {
wcb = new ExistsWatchRegistration(watcher, clientPath);
}
final String serverPath = prependChroot(clientPath);
//请求头
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.exists);
//请求对象
ExistsRequest request = new ExistsRequest();
request.setPath(serverPath); //first
request.setWatch(watcher != null); //true/false
//response返回对象
SetDataResponse response = new SetDataResponse();
// cnxn 网络通信的负责处理的类
ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
if (r.getErr() != 0) {//返回的错误码去判断返回的结果
if (r.getErr() == KeeperException.Code.NONODE.intValue()) {
return null;
}
throw KeeperException.create(KeeperException.Code.get(r.getErr()), clientPath);
}
//返回stata
return response.getStat().getCzxid() == -1 ? null : response.getStat();
}
---------------------------------------------------------------------------------------
public ReplyHeader submitRequest(
RequestHeader h,
Record request,
Record response,
WatchRegistration watchRegistration,
WatchDeregistration watchDeregistration) throws InterruptedException {
ReplyHeader r = new ReplyHeader();
//构建Packet 数据包，主要是需要传递的内容(queuePacket 实际上在这里是要将一个数据包发送到队列中，这符合我么图解中说明的那样)
Packet packet = queuePacket(
h,
r,
request,
response,
null,
null,
null,
null,
watchRegistration,
watchDeregistration);
synchronized (packet) {// 加锁
if (requestTimeout > 0) { //如果携带了请求超时时间. 带超时时间的等待
// Wait for request completion with timeout
waitForPacketFinish(r, packet);
} else {
// Wait for request completion infinitely
while (!packet.finished) { //只要packet没有处理完成，那么一直调用wait等待。
packet.wait(); //阻塞
}
}
}
if (r.getErr() == Code.REQUESTTIMEOUT.intValue()) {
sendThread.cleanAndNotifyState();
}
return r;
}
---------------------------------------------------------------------------------------
public Packet queuePacket(
RequestHeader h,
ReplyHeader r,
Record request,
Record response,
AsyncCallback cb,
String clientPath,
String serverPath,
Object ctx,
WatchRegistration watchRegistration,
WatchDeregistration watchDeregistration) {
Packet packet = null;
// Note that we do not generate the Xid for the packet yet. It is
// generated later at send-time, by an implementation of ClientCnxnSocket::doIO(),
// where the packet is actually sent.
packet = new Packet(h, r, request, response, watchRegistration);
packet.cb = cb;
packet.ctx = ctx;
packet.clientPath = clientPath;
packet.serverPath = serverPath;
packet.watchDeregistration = watchDeregistration;
// The synchronized block here is for two purpose:
// 1. synchronize with the final cleanup() in SendThread.run() to avoid race
// 2. synchronized against each packet. So if a closeSession packet is added,
// later packet will be notified.
synchronized (state) {
if (!state.isAlive() || closing) {
conLossPacket(packet);
} else {
// If the client is asking to close the session then
// mark as closing
if (h.getType() == OpCode.closeSession) {
closing = true;
}
//添加到阻塞队列
outgoingQueue.add(packet);
}
}
// 唤醒处于阻塞在selector.select上的线程
// 那么sendThread是在哪里初始化的呢？
sendThread.getClientCnxnSocket().packetAdded();
return packet;
}
---------------------------------------------------------------------------------------
//在zookeeper的构造方法里面
public ZooKeeper(
String connectString,
int sessionTimeout,
Watcher watcher,
boolean canBeReadOnly,
HostProvider aHostProvider) throws IOException {
this(connectString, sessionTimeout, watcher, canBeReadOnly, aHostProvider, null);
}
public ZooKeeper(
.........
// 客户端和服务端的一个连接
cnxn = createConnection(
connectStringParser.getChrootPath(),
hostProvider,
sessionTimeout,
this,
watchManager,
getClientCnxnSocket(),
canBeReadOnly);
cnxn.start();
}
public void start() {
//发送线程
sendThread.start();
//事件线程（触发事件的线程，
// 也就是说当服务端触发了事件通知到客户端之后，客户端需要从本地的事件列表中去读取watcher，并且进行回调）
eventThread.start();
}
// 后续就是一堆nio netty等流程的内容，暂不关注
---------------------------------------------------------------------------------------
public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
// We have the request, now process and setup for next
InputStream bais = new ByteBufferInputStream(incomingBuffer);
BinaryInputArchive bia = BinaryInputArchive.getArchive(bais);
RequestHeader h = new RequestHeader();
h.deserialize(bia, "header"); //反序列化header
cnxn.incrOutstandingAndCheckThrottle(h);
incomingBuffer = incomingBuffer.slice();
//根据请求类型进行不同的处理
if (h.getType() == OpCode.auth) {
// 授权
} else if (h.getType() == OpCode.sasl) {
processSasl(incomingBuffer, cnxn, h);
} else {
if (shouldRequireClientSaslAuth() && !hasCnxSASLAuthenticated(cnxn)) {
ReplyHeader replyHeader = new ReplyHeader(h.getXid(), 0, Code.SESSIONCLOSEDREQUIRESASLAUTH.intValue());
cnxn.sendResponse(replyHeader, null, "response");
cnxn.sendCloseSession();
cnxn.disableRecv();
} else {
Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(), h.getType(), incomingBuffer, cnxn.getAuthInfo());
int length = incomingBuffer.limit();
if (isLargeRequest(length)) {
// checkRequestSize will throw IOException if request is rejected
checkRequestSizeWhenMessageReceived(length);
si.setLargeRequestSize(length);
}
si.setOwner(ServerCnxn.me);
submitRequest(si); //提交请求(异步有关系)
}
}
}
public void submitRequest(Request si) {
enqueueRequest(si);
}
public void enqueueRequest(Request si) {
// 有点类似限流的逻辑
if (requestThrottler == null) {
synchronized (this) {
try {
while (state == State.INITIAL) {
wait(1000);
}
} catch (InterruptedException e) {
LOG.warn("Unexpected interruption", e);
}
if (requestThrottler == null) {
throw new RuntimeException("Not started");
}
}
}
requestThrottler.submitRequest(si);
}
public void submitRequest(Request request) {
if (stopping) {//如果服务端在终止的过程，则删除这个请求
LOG.debug("Shutdown in progress. Request cannot be processed");
dropRequest(request);
} else {
submittedRequests.add(request);
}
}
---------------------------------------------------------------------------------------
// 此时终于找到了一直轮询的线程了
public void run() {
try {
while (true) {
if (killed) {
break;
}
// 到这里 一个典型的生产者消费者的方式才很清晰
Request request = submittedRequests.take();
if (Request.requestOfDeath == request) {
break;
}
if (request.mustDrop()) {
continue;
}
// Throttling is disabled when maxRequests = 0
//节流阀是否处于关闭状态，=0表示关闭
if (maxRequests > 0) {
while (!killed) {
if (dropStaleRequests && request.isStale()) {
// Note: this will close the connection
dropRequest(request);
ServerMetrics.getMetrics().STALE_REQUESTS_DROPPED.add(1);
request = null;
break;
}
//限流动作
if (zks.getInProcess() < maxRequests) {
break;
}
//等待.
throttleSleep(stallTime);
}
}
if (killed) {
break;
}
// A dropped stale request will be null
if (request != null) {
if (request.isStale()) {
ServerMetrics.getMetrics().STALE_REQUESTS.add(1);
}
zks.submitRequestNow(request);
}
}
} catch (InterruptedException e) {
LOG.error("Unexpected interruption", e);
}
int dropped = drainQueue();
LOG.info("RequestThrottler shutdown. Dropped {} requests", dropped);
}
public void submitRequestNow(Request si) {
if (firstProcessor == null) {
synchronized (this) {
try {
// Since all requests are passed to the request
// processor it should wait for setting up the request
// processor chain. The state will be updated to RUNNING
// after the setup.
while (state == State.INITIAL) {
wait(1000);
}
} catch (InterruptedException e) {
LOG.warn("Unexpected interruption", e);
}
if (firstProcessor == null || state != State.RUNNING) {
throw new RuntimeException("Not started");
}
}
}
try {
touch(si.cnxn);
boolean validpacket = Request.isValid(si.type);
if (validpacket) { //如果packet合法
setLocalSessionFlag(si);
//通过一个处理器链来处理这个请求
// PrepRequestProcessor(SyncRequestProcessor(FinalRequestProcessor))
//
firstProcessor.processRequest(si);
---------------------------------------------------------------------------------------
//构建一个请求处理链路
//单机环境的处理链路：PrepRequestProcessor(SyncRequestProcessor(FinalRequestProcessor))
protected void setupRequestProcessors() {
//最终的处理器
RequestProcessor finalProcessor = new FinalRequestProcessor(this);
//SyncRequestProcessor 同步处理器 将数据同步到本地磁盘
RequestProcessor syncProcessor = new SyncRequestProcessor(this, finalProcessor);
// 同步处理器最终会有一个 写入到快照文件，也就是需要设置自己的磁盘同步策略
// 其实就是性能 和 一致性的取舍问题
((SyncRequestProcessor) syncProcessor).start();
//PrepRequestProcessor 预处理器
firstProcessor = new PrepRequestProcessor(this, syncProcessor);
((PrepRequestProcessor) firstProcessor).start();
}
---------------------------------------------------------------------------------------
if (si.cnxn != null) {
incInProcess();
}
} else {
LOG.warn("Received packet at server of unknown type {}", si.type);
// Update request accounting/throttling limits
requestFinished(si);
new UnimplementedRequestProcessor().processRequest(si);
}
} catch (MissingSessionException e) {
LOG.debug("Dropping request.", e);
// Update request accounting/throttling limits
requestFinished(si);
} catch (RequestProcessorException e) {
LOG.error("Unable to process request", e);
// Update request accounting/throttling limits
requestFinished(si);
}
}
---------------------------------------------------------------------------------------
case OpCode.exists: {
lastOp = "EXIS";
// TODO we need to figure out the security requirement for this!
ExistsRequest existsRequest = new ExistsRequest();
ByteBufferInputStream.byteBuffer2Record(request.request, existsRequest);
path = existsRequest.getPath();
if (path.indexOf('\0') != -1) {
throw new KeeperException.BadArgumentsException();
}
//通过zk得到stat
// 从这里可以看出来前面图解中说到的HashMap 里面的 Set<Watcher> 中存储的watcher其实是网络对象
// 之所以这样去实现当path发生变化的时候，需要告诉所有的监视者，记住这个网络连接将数据返回出去就行了。
Stat stat = zks.getZKDatabase().statNode(path, existsRequest.getWatch() ? cnxn : null);
rsp = new ExistsResponse(stat);
requestPathMetricsCollector.registerRequest(request.type, path);
break;
}
public Stat statNode(String path, Watcher watcher) throws KeeperException.NoNodeException {
Stat stat = new Stat();
DataNode n = nodes.get(path);
// 到这里才到图解中注册的流程
if (watcher != null) {//服务端的注册的流程
dataWatches.addWatch(path, watcher);
}
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
n.copyStat(stat);
}
updateReadStat(path, 0L);
return stat;
}
public boolean addWatch(String path, Watcher watcher) {
/**
* watcher 表示当前的一个注册监听的一个连接
* path 表示监听的路径
*/
return addWatch(path, watcher, WatcherMode.DEFAULT_WATCHER_MODE);
}
STANDARD(false, false),
PERSISTENT(true, false),
PERSISTENT_RECURSIVE(true, true)
;
public static final WatcherMode DEFAULT_WATCHER_MODE = WatcherMode.STANDARD;
// 接下来就是自然而然的保存了
// 表示节点到watcher集合的映射
private final Map<String, Set<Watcher>> watchTable = new HashMap<>();
// 表示从watcher到所有节点的映射
private final Map<Watcher, Set<String>> watch2Paths = new HashMap<>();
public synchronized boolean addWatch(String path, Watcher watcher, WatcherMode watcherMode) {
if (isDeadWatcher(watcher)) {
LOG.debug("Ignoring addWatch with closed cnxn");
return false;
}
Set<Watcher> list = watchTable.get(path);
if (list == null) {
// don't waste memory if there are few watches on a node
// rehash when the 4th entry is added, doubling size thereafter
// seems like a good compromise
list = new HashSet<>(4);
// 保存
watchTable.put(path, list);
}
list.add(watcher);
Set<String> paths = watch2Paths.get(watcher);
if (paths == null) {
// cnxns typically have many watches, so use default cap here
paths = new HashSet<>();
watch2Paths.put(watcher, paths);
}
//设置监听模式
watcherModeManager.setWatcherMode(watcher, path, watcherMode);
return paths.add(path);
}
// 此时就完成了整个服务端的一个保存
---------------------------------------------------------------------------------------
// 在SendThread中 有一个叫readResponse 的方法
void readResponse(ByteBuffer incomingBuffer) throws IOException {
ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);
BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
ReplyHeader replyHdr = new ReplyHeader();
replyHdr.deserialize(bbia, "header");
switch (replyHdr.getXid()) {
case PING_XID:
......
case AUTHPACKET_XID:
......
case NOTIFICATION_XID:
......
default:
break;
}
// If SASL authentication is currently in progress, construct and
// send a response packet immediately, rather than queuing a
// response as with other packets.
if (tunnelAuthInProgress()) {
GetSASLRequest request = new GetSASLRequest();
request.deserialize(bbia, "token");
zooKeeperSaslClient.respondToServer(request.getToken(), ClientCnxn.this);
return;
}
Packet packet;
synchronized (pendingQueue) {
if (pendingQueue.size() == 0) {
throw new IOException("Nothing in the queue, but got " + replyHdr.getXid());
}
packet = pendingQueue.remove();
}
/*
* Since requests are processed in order, we better get a response
* to the first request!
*/
try {
if (packet.requestHeader.getXid() != replyHdr.getXid()) {
packet.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());
throw new IOException("Xid out of order. Got Xid " + replyHdr.getXid()
+ " with err " + replyHdr.getErr()
+ " expected Xid " + packet.requestHeader.getXid()
+ " for a packet with details: " + packet);
}
packet.replyHeader.setXid(replyHdr.getXid());
packet.replyHeader.setErr(replyHdr.getErr());
packet.replyHeader.setZxid(replyHdr.getZxid());
if (replyHdr.getZxid() > 0) {
lastZxid = replyHdr.getZxid();
}
if (packet.response != null && replyHdr.getErr() == 0) {
packet.response.deserialize(bbia, "response");
}
LOG.debug("Reading reply session id: 0x{}, packet:: {}", Long.toHexString(sessionId), packet);
} finally {
finishPacket(packet);
}
}
在zookeeper.class中
public void register(int rc) {
if (shouldAddWatch(rc)) {//如果服务端已经建立了映射关系，则需要在客户端建立好关系
Map<String, Set<Watcher>> watches = getWatches(rc);
synchronized (watches) {
Set<Watcher> watchers = watches.get(clientPath);
if (watchers == null) {
watchers = new HashSet<Watcher>();
watches.put(clientPath, watchers);
}
watchers.add(watcher);
}
}
}
/*
后续的流程，如果服务端返回成功的话，那么就保存好了，此时关系就建立好了，而触发监听的方式。
服务端在一个地方发现数据发生变更的时候，直接在服务端找到一个对应的watcher，去推送消息就行了，客户端收到消息，判断消息类型，根据映射关系去找到watcher回调即可。
*/
