zookeeper的ZAB协议的原理以及底层源码实现超级详解
Posted huisheng_qaq
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了zookeeper的ZAB协议的原理以及底层源码实现超级详解相关的知识，希望对你有一定的参考价值。
zookeeper的ZAB协议的原理以及实现
一，zookeeper的ZAB协议
1，ZAB概述
ZAB：zookeeper atomic broadcast(zookeeper原子广播协议)
ZAB协议主要包括这个 原子广播 和 崩溃恢复
原子广播
就是说集群的主结点leader用来写，其他follow从结点只用来读。在主结点写完会将数据同步到从结点，只要写入成功的从结点的数量超过一半，那么这个数据就同步成功。这个主结点同步到从结点可能会有一定的延迟，因此这个zookeeper主要是为了保证这个数据的最终一致性，也可以叫为顺序一致性。
崩溃恢复
如果在主结点刚把数据写完，这个主结点挂了，那么这个集群就会重新选举新的leader。选leader的规则就是先比较zxid事务id，再比较这个机器对应的myid，谁大谁被选为leader。
二，ZAB协议流程的源码实现
需要下载zookeeper源码，可以参考上一篇：https://blog.csdn.net/zhenghuishengq/article/details/126673923?spm=1001.2014.3001.5502
1，客户端建立连接
1，先创建一个zookeeper对象
ZooKeeper zooKeeper=new ZooKeeper(...);
这个zookeeper的构造方法可能如下，里面会有很多参数，里面主要是会对创建一个connection的连接。
public ZooKeeper(
String connectString,
int sessionTimeout,
Watcher watcher,
long sessionId,
byte[] sessionPasswd,
boolean canBeReadOnly,
HostProvider hostProvider,
ZKClientConfig clientConfig) throws IOException
//增加一个监听机制
validateWatcher(watcher);
this.clientConfig = clientConfig != null ? clientConfig : new ZKClientConfig();
ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
this.hostProvider = hostProvider;
//建立一个ClientCnxn连接，会和这个服务端建立连接
cnxn = new ClientCnxn(
connectStringParser.getChrootPath(),
hostProvider,
sessionTimeout,
this.clientConfig,
watcher,
getClientCnxnSocket(),
sessionId,
sessionPasswd,
canBeReadOnly);
//开始连接
cnxn.start();
2，然后进入这个开始连接的start方法，这个方法里面主要是会去开启两个线程，一个线程主要用来连接服务端，一个线程主要用来响应连接后的事件
public void start()
sendThread.start();
eventThread.start();
开启这个了线程之后，主要是查看这个线程的run方法。
接下来看第一个 sendThread 线程的run方法，主要如下，用于连接这个服务端，主要是基于nio和netty两种方式实现这个连接。在连接成功之后，会通过这个nio轮询的方式监听里面的读写事件的发生，并对这个事件进行处理
@Override
public void run()
while (state.isAlive())
try
//如果没有建立连接
if (!clientCnxnSocket.isConnected())
onConnecting(serverAddress);
//建立连接
startConnect(serverAddress);
//在建立连接之后，会通过这个nio监听这个读写事件并处理
clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);
private void startConnect(InetSocketAddress addr) throws IOException
//基于这个nio或者netty两种方式实现，连接这个客户端
clientCnxnSocket.connect(addr);
接下来看第二个线程 eventThread 的run方法，主要是调用一个watcher的一个监听机制
@Override
public void run()
while (true)
//从阻塞队列里面获取一个事件对象
Object event = waitingEvents.take();
//处理这个事件
processEvent(event);
private void processEvent(Object event)
WatcherSetEventPair pair = (WatcherSetEventPair) event;
//Watcher监听机制，用于事件回调
watcher.process(pair.event);
2，客户端写数据
3，建立连接成功之后，就可以开始写数据的操作，主要是通过这个create的这个方法实现
zooKeeper.create("/myconfig", bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
这个create的方法具体如下，就是会将这个发送的数据以及存放的路径做一个封装
public String create(
final String path,
byte[] data,
List<ACL> acl,
CreateMode createMode) throws KeeperException, InterruptedException
//会将传进来的数据和路径封装到一个request里面
request.setData(data);
request.setFlags(createMode.toFlag());
request.setPath(serverPath);
//通过这个连接对象实现这个客户端向服务端发送数据
ReplyHeader r = cnxn.submitRequest(h, request, response, null);
发送数据的具体实现如下，会对这个传来的数据进行一个打包的操作，在数据打包之后，会将这个数据存放到一个阻塞队列里面
public ReplyHeader submitRequest(...)
//对这个传过来的数据进行一个打包操作
Packet packet = queuePacket(h,r,request,response,null,null,null,null,
watchRegistration,watchDeregistration);
waitForPacketFinish(r, packet);
public Packet queuePacket()
synhronized (outgoingQueue)
if (!state.isAlive() || closing)
conLossPacket(packet);
else
if (h.getType() == OpCode.closeSession)
closing = true;
//将打包的数据存放到一个阻塞队列里面
outgoingQueue.add(packet);
//唤醒被阻塞的selector，然后向这个管道写入一个数据，
//这样就可以触发前面的sendThread线程，并触发里面的写事件，将数据写入到服务端
sendThread.getClientCnxnSocket().packetAdded();
最终会通过这个Socket的write写入事件，将这个序列化后的数据存放到buffer里面，然后通过这个SocketChannel的write方法将数据写入到服务端。数据主要是通过这个outgoingQueue队列，以异步的方式将这个信息发送到服务端。
3，服务端接收数据
4，服务端这边主要是在这个 ServerCnxnFactory 类下面的 createFactory 方法里面来构建与客户端的连接，这个服务端接收数据主要是通过主结点和这个客户端进行一个交互
public abstract class ServerCnxnFactory
static public ServerCnxnFactory createFactory() throws IOException
//这里主要是选择nio的方式或者选择netty的方式建立一个连接
String serverCnxnFactoryName =
System.getProperty(ZOOKEEPER_SERVER_CNXN_FACTORY);
if (serverCnxnFactoryName == null)
serverCnxnFactoryName = NioserverCnxnFactory.class.getName();
try
//然后通过这个反射的方式找到对应的server，如使用的netty连接就会找nettyServer
ServerCnxnFactory serverCnxnFactory = (ServerCnxnFactory) Class.forName(serverCnxnFactoryName)
.getDeclaredConstructor().newInstance();
return serverCnxnFactory;
5，接下来主要查看这个 NettyServerCnxnFactory 类的这个构造方法，底层主要是一些netty的一些实现逻辑。主要用来实现数据的传输
NettyServerCnxnFactory()
EventLoopGroup bossGroup = NettyUtils.newNioOrEpollEventLoopGroup(
NettyUtils.getClientReachableLocalInetAddressCount());
EventLoopGroup workerGroup = NettyUtils.newNioOrEpollEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap()
.group(bossGroup, workerGroup)
.channel(NettyUtils.nioOrEpollServerSocketChannel())
// parent channel options
.option(ChannelOption.SO_REUSEADDR, true)
// child channels options
.childOption(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_LINGER, -1)
.childHandler(new ChannelInitializer<SocketChannel>()
@Override
protected void initChannel(SocketChannel ch) throws Exception
ChannelPipeline pipeline = ch.pipeline();
if (secure)
initSSL(pipeline, false);
else if (shouldUsePortUnification)
initSSL(pipeline, true);
pipeline.addLast("servercnxnfactory", channelHandler);
);
this.bootstrap = configureBootstrapAllocator(bootstrap);
this.bootstrap.validate();
6，在建立好连接之后，服务端这边会通过这个channelRead 这个方法来读取通道的信息，就是客户端发过来的信息。
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
try
if (LOG.isTraceEnabled())
LOG.trace("message received called ", msg);
try
if (LOG.isDebugEnabled())
LOG.debug("New message from ", msg, ctx.channel());
//服务端的连接对象
NettyServerCnxn cnxn = ctx.channel().attr(CONNECTION_ATTRIBUTE).get();
if (cnxn == null)
else
//处理客户端传过来的数据
cnxn.processMessage((ByteBuf) msg);
catch (Exception ex)
LOG.error("Unexpected exception in receive", ex);
throw ex;
finally
ReferenceCountUtil.release(msg);
//处理这个客户端传过来的数据
void processMessage(ByteBuf buf)
receiveMessage(buf);
7，接下来就是正式的接收这个发送过来的数据，并对这个数据进行处理
private void receiveMessage(ByteBuf message)
//读取message，将数据读取到服务端的byteBuffer里面
message.readBytes(bb);
//处理这个打包好的数据
zks.processPacket(this, bb);
接下来进入这个processPacket 这个方法，主要是用来解析打包的数据。并且在服务端中，又会将这个数据进行一个打包，封装到一个request的一个阻塞队列里面，最后将这个打包的数据进行提交
public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException
//二进制流接收数据
InputStream bais = new ByteBufferInputStream(incomingBuffer);
BinaryInputArchive bia = BinaryInputArchive.getArchive(bais);
//将传送过来的序列化的数据进行一个反序列化
RequestHeader h = new RequestHeader();
h.deserialize(bia, "header");
//最后将客户端传过来的数据又封装到一个request的对象里面
Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(),
h.getType(), incomingBuffer, cnxn.getAuthInfo());
si.setOwner(ServerCnxn.me);
setLocalSessionFlag(si);
//提交这个数据
submitRequest(si);
4，服务端主结点处理数据
8，就是进入上面的这个 submitRequest 的提交数据的这个方法，里面主要是通过一个processRequest的这个方法来进行处理这些数据
public void submitRequest(Request si)
firstProcessor.processRequest(si);
9，这个firstProcessor的处理器主要是在 LeaderZooKeeperServer 类下面的这个 setupRequestProcessors 的这个方法初始化的。服务端这边主要会初始化这个Processor的一个链条，底层主要是通过这个责任链的方式实现。责任链主要是为了分工合作，模块解耦
@Override
protected void setupRequestProcessors()
RequestProcessor finalProcessor = new FinalRequestProcessor(this);
//将上一个processor放入下一个processor，开始构建一个责任链的一个链条
RequestProcessor toBeAppliedProcessor = new Leader.ToBeAppliedRequestProcessor(finalProcessor, getLeader());
commitProcessor = new CommitProcessor(toBeAppliedProcessor,
Long.toString(getServerId()), false,
getZooKeeperServerListener());
commitProcessor.start();
ProposalRequestProcessor proposalProcessor = new ProposalRequestProcessor(this,
commitProcessor);
proposalProcessor.initialize();
prepRequestProcessor = new PrepRequestProcessor(this, proposalProcessor);
//获取队列的数据，并对数据进行读取
prepRequestProcessor.start();
firstProcessor = new LeaderRequestProcessor(this, prepRequestProcessor);
setupContainerManager();
这个链条如下
10，在执行这个链条的过程中，会通过这个prepRequestProcessor这个线程来读取加在服务端队列的里面的消息，接下来主要就是查看这个线程里面的run方法。这个结点其主要是为了填充这个zxid，就是事务id
@Override
public void run()
try
Request request = submittedRequests.take();
//通过这个方法进行最终的处理
pRequest(request);
在这个pRequest方法里面，会比较之前客户端传过来的命令，比如说create，delete命令等，那么服务端就会执行具体的操作。
protected void pRequest(Request request) throws RequestProcessorException
try
switch (request.type)
case OpCode.createContainer:
case OpCode.create:
case OpCode.create2:
CreateRequest create2Request = new CreateRequest();
//创建命令的具体逻辑
pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true);
break;
case OpCode.createTTL:
CreateTTLRequest createTtlRequest = new CreateTTLRequest();
pRequest2Txn(request.type, zks.getNextZxid(), request, createTtlRequest, true);
break;
case OpCode.deleteContainer:
case OpCode.delete:
DeleteRequest deleteRequest = new DeleteRequest();
pRequest2Txn(request.type, zks.getNextZxid(), request, deleteRequest, true);
break;
case OpCode.setData:
SetDataRequest setDataRequest = new SetDataRequest();
pRequest2Txn(request.type, zks.getNextZxid(), request, setDataRequest, true);
break;
...
接下来就是主要查看一下这个创建命令，主要是通过这个 pRequest2Txn 方法实现，并且里面的这个参数zxid，是一个automic的一个原子类型，不存在这个线程安全问题。获取的命令越新，那么这个zxid的值就越大。 每从这个内存队列里面获取一条消息，那么这个zxid的值就会加1
protected void pRequest2Txn(int type, long zxid, Request request,
Record record, boolean deserialize)
throws KeeperException, IOException, RequestProcessorException
//将这个zxid填充回request
request.zxid = zks.getZxid();
//由下一个process处理
nextProcessor.processRequest(request);
5，主结点同步数据到从结点(ZAB协议)
5.1，发送这个propose（第一阶段）
11，完成了这个链条中的第二个环节之后，就进入第三个环节，即ProposalRequestProcessor的这个结点。这一环节只要是为了同步数据到从结点，并且将数据同步到从结点之后，会将这个数据在本地磁盘里面保存一份
public class ProposalRequestProcessor implements RequestProcessor
//主要是会走这个方法
public void processRequest(Request request) throws RequestProcessorException
nextProcessor.processRequest(request);
//propose处理这个request
zks.getLeader().propose(request);
//将数据写入到本地磁盘
syncProcessor.processRequest(request);
接下来查看这个propose方法，会对主结点中的数据进行一个预处理，并将数据发送给全部的
以上是关于zookeeper的ZAB协议的原理以及底层源码实现超级详解的主要内容，如果未能解决你的问题，请参考以下文章