ZAB：zookeeper atomic broadcast(zookeeper原子广播协议)
ZAB协议主要包括这个 原子广播 和 崩溃恢复
原子广播
就是说集群的主结点leader用来写，其他follow从结点只用来读。在主结点写完会将数据同步到从结点，只要写入成功的从结点的数量超过一半，那么这个数据就同步成功。这个主结点同步到从结点可能会有一定的延迟，因此这个zookeeper主要是为了保证这个数据的最终一致性，也可以叫为顺序一致性。
崩溃恢复
如果在主结点刚把数据写完，这个主结点挂了，那么这个集群就会重新选举新的leader。选leader的规则就是先比较zxid事务id，再比较这个机器对应的myid，谁大谁被选为leader。
需要下载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);
}
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);
}
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);
}
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方法，会对主结点中的数据进行一个预处理，并将数据发送给全部的从结点
public Proposal propose(Request request) throws XidRolloverException {
//序列化
byte[] data = SerializeUtils.serializeRequest(request);
proposalStats.setLastBufferSize(data.length);
//对数据进行打包，里面会有几种数据类型，如ping，ack等
QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
//将这些数据全部发送出去
sendPacket(pp);
}
这个sendPacket方法，就是会轮询的将数据发送给所有的这个follow从结点
void sendPacket(QuorumPacket qp) {
synchronized (forwardingFollowers) {
//循环发送
for (LearnerHandler f : forwardingFollowers) {
f.queuePacket(qp);
}
}
}
数据发送完同时也会将数据存放在这个本地磁盘里面，主要是通过这个SyncRequestProcessor 类里面的这个processRequest线程实现，主要查看这个线程的run方法。就是leader主结点将数据存在本地磁盘
@Override
public void run() {
//将一些数据初始化到磁盘上面
zks.getZKDatabase().rollLog();
//调用一个flush方法，主要用于写日志文件
//主要是写一些事物文件和快照文件
flush(toFlush);
}
12，主要是通过这个 AckRequestProcessor 类实现，里面有一个processRequest的这个方法，就是首先这个leader会先给自己发一个ack，这样在后面统计这个只有从结点的响应的ack的同时，还需要加上这个主结点的ack。
public void processRequest(Request request) {
QuorumPeer self = leader.self;
if(self != null)
leader.processAck(self.getId(), request.zxid, null);
else
LOG.error("Null QuorumPeer");
}
通过这个processAck 方法可以知道，最终会将这个结点的sid存到主结点的一个hashset的一个集合里面。
//将这台机器的sid存在这个hashset里面
p.addAck(sid);
//尝试判断这个票数是否大于一半，大于一半则提交
boolean hasCommitted = tryToCommit(p, zxid, followerAddr);
13，主结点会给所有的从结点发送数据，会和这些从结点建立nio的一个连接，然后通过这个 LearnerHandler 类来发送这个消息。这个类继承了一个线程类，那么主要看这个类的run方法
public class LearnerHandler extends ZooKeeperThread {
@Override
public void run() {
//会开始发送这个数据包
startSendingPackets();
}
private void startSendingPackets() throws InterruptedException {
public void run() {
try {
//开始发送数据包
sendPackets();
} catch (InterruptedException e) {
LOG.warn("Unexpected interruption " + e.getMessage());
}
}
}.start();
}
private void sendPackets() throws InterruptedException {
QuorumPacket p;
//从队列中获取数据
p = queuedPackets.poll();
//通过bio的方式，将序列化的数据写入到从结点
oa.writeRecord(p, "packet");
}
}
14，主结点将消息发送到这个从结点之后，在这个Follow的这个类里面，通过这个followLeader方法来读取主结点的发过来的消息，同时也会将这个数据存储在这个本地磁盘里面
public class Follower extends Learner{
void followLeader() throws InterruptedException {
while (this.isRunning()) {
//读取传过来的数据
readPacket(qp);
//处理这个packet的这个数据包
processPacket(qp);
}
}
}
从结点处理主结点发送的这个数据包的具体实现如下
//处理这个数据包的过程如下
protected void processPacket(QuorumPacket qp) throws Exception{
case Leader.PING: ping(qp); break;
//将传过来的数据写入到磁盘
case Leader.PROPOSAL: fzk.logRequest(hdr, txn); break;
case Leader.COMMIT: fzk.commit(qp.getZxid()); break;
...
}
15，从结点在处理完数据之后，会通过这个 SendAckRequestProcessor 类里面的 processRequest 方法来给这个主结点返回一个ack
public class SendAckRequestProcessor implements RequestProcessor, Flushable {
public void processRequest(Request si) {
//构建一个返回一个ACK的一个数据包
QuorumPacket qp = new QuorumPacket(Leader.ACK, si.getHdr().getZxid(), null,null);
//将数据写入,通过这个bio的连接，将数据写会给这个主结点
learner.writePacket(qp, false);
}
}
16，从结点同步完数据之后，会返回一个ack的一个确认机制，主结点主要是 LearnerHandler 线程类的run方法里面实现，里面有一个while循环一直接收这个从结点发的消息。类型为ack时会和之前的流程一样，将这个从结点的sid存放在一个hashset的一个集合里面，最后会去尝试这个commit提交，大于一半就会提交
public class LearnerHandler extends ZooKeeperThread {
@Override
public void run() {
while (true) {
//leader获取数据
qp = new QuorumPacket();
ia.readRecord(qp, "packet");
//获取数据的类型
switch (qp.getType()) {
//这个ack会走和主结点ack一样的流程
case Leader.ACK: ...; break;
case Leader.PING: ...; break;
case Leader.REVALIDATE: ...; break;
case Leader.REQUEST: ...; break;
}
}
}
}
17，在主结点获取到这个ack之后，都会有一个尝试commit的提交 *** 作，如果这个票数过半，那么就会走这个正式的commit的提交 *** 作。就是说leader会再发起一个请求，告诉这些从结点也可以进行数据的提交，就是将之前存在日志里面的数据加载到内存里面，那么其他客户端来查询就可以从这个从结点里面查出这个数据。从结点数据提交之后，这个主结点的数据也会提交。
synchronized public boolean tryToCommit(Proposal p, long zxid, SocketAddress followerAddr) {
commit(zxid);
inform(p);
//从结点提交之后，主结点这边也会进行一个提交
zk.commitProcessor.commit(p.request);
}
然后主要查看这个commit 方法
public void commit(long zxid) {
synchronized(this){
lastCommitted = zxid;
}
//又会构建一个数据包，这个类型是COMMIT类型，同时返回一个zxid
QuorumPacket qp = new QuorumPacket(Leader.COMMIT, zxid, null, null);
//轮询的方式发送给所有的从结点，
sendPacket(qp);
}
void sendPacket(QuorumPacket qp) {
//轮询的方式发送
synchronized (forwardingFollowers) {
for (LearnerHandler f : forwardingFollowers) {
f.queuePacket(qp);
}
}
}
再查看这个inform 方法，可以发现这个数据也会同步给Observer的结点
public void inform(Proposal proposal) {
QuorumPacket qp = new QuorumPacket(Leader.INFORM, proposal.request.zxid,
proposal.packet.getData(), null);
sendObserverPacket(qp);
}
在从结点提交完数据之后，主结点也会提交数据，就是将存在磁盘里面的数据加载到内存里面
protected void processCommitted() {
Request request;
//从队列中获取消息
request = committedRequests.poll();
Request pending = nextPending.get();
sendToNextProcessor(pending);
}
18，接下来进入链条的倒数第二个结点ToBeAppliedRequestProcessor
static class ToBeAppliedRequestProcessor implements RequestProcessor{
public void processRequest(Request request) throws RequestProcessorException {
//责任链模式，直接进入下一个request
next.processRequest(request);
}
}
19，那么直接进入责任链里面的最后一个节点 FinalRequestProcessor
public class FinalRequestProcessor implements RequestProcessor {
public void processRequest(Request request) {
//处理事务
rc = zks.processTxn(request);
}
}
接下来可以查看这个processTxn的这个方法，就是将这个内存中的数据存储到对应的树形结构里面
private ProcessTxnResult processTxn(Request request, TxnHeader hdr,Record txn) {
rc = getZKDatabase().processTxn(hdr, txn);
}
public ProcessTxnResult processTxn(TxnHeader hdr, Record txn) {
//将结果加入到zookeeper的树形结构中
return dataTree.processTxn(hdr, txn);
}
20，依旧是在这个责任链模式的最后一个结点FinalRequestProcessor，里面会有一个服务端给客户端的响应。就是告知客户端这条命令执行是否成功失败
public class FinalRequestProcessor implements RequestProcessor {
public void processRequest(Request request) {
//给客户端响应
cnxn.sendResponse(hdr, rsp, "response");
}
}
然后就是查看这个sendResponse方法，
public void sendResponse(ReplyHeader h, Record r, String tag) throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
//对数据进行序列化
BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
try {
baos.write(fourBytes);
bos.writeRecord(h, "header");
if (r != null) {
bos.writeRecord(r, tag);
}
baos.close();
} catch (IOException e) {
LOG.error("Error serializing response");
}
byte b[] = baos.toByteArray();
serverStats().updateClientResponseSize(b.length - 4);
ByteBuffer bb = ByteBuffer.wrap(b);
bb.putInt(b.length - 4).rewind();
//以流的方式发送
sendBuffer(bb);
}
在看这个sendBuffer方法，将封装给客户端的数据返回
@Override
public void sendBuffer(ByteBuffer sendBuffer) {
if (sendBuffer == ServerCnxnFactory.closeConn) {
close();
return;
}
channel.writeAndFlush(Unpooled.wrappedBuffer(sendBuffer)).addListener(
onSendBufferDoneListener);
}
21，由于这个客户端和这个服务一开始就建立了这个nio连接或者netty连接，因此在服务端给客户端发送这个数据的时候，客户端这边也可以立马收到响应。
依旧是在这个ClientCnxn类里面，找到这个这个客户端的doTransport方法，就是会去处理对应的事件
public class ClientCnxn {
clientCnxnSocket.doTransport(to, pendingQueue, ClientCnxn.this);
}
在nio的这个doTransport方法里面，会去判断这个事件的读写
@Override
void doTransport(int waitTimeOut, List<Packet> pendingQueue, ClientCnxn cnxn) throws IOException, InterruptedException {
//判断是读事件开始写事件
if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
doIO(pendingQueue, cnxn);
}
}
22，客户端这边主要是通过这个doIO方法来读取这个io流的数据
@Override
void doIO(int waitTimeOut, List<Packet> pendingQueue, ClientCnxn cnxn) throws IOException, InterruptedException {
//读取这个io
sendThread.readResponse(incomingBuffer);
}
void readResponse(ByteBuffer incomingBuffer) throws IOException {
//客户端这边会有一个watcher的监听器
WatchedEvent we = new WatchedEvent(event);
//会将这个事件加入到队列中
eventThread.queueEvent( we );
}
接下来查看这个重点的queueEvent方法，就是会将这个事件加入到一个阻塞队列里面。在zookeeper客户端启动的时候，就会创建两个线程，一个就是用于监听机制的eventThread线程，监听的事件就是现在加入的事件。
private void queueEvent(WatchedEvent event,Set<Watcher> materializedWatchers) {
WatcherSetEventPair pair = new WatcherSetEventPair(watchers, event);
waitingEvents.add(pair);
}
一个zookeeper的原子消息的协议，主要通过两阶段提交的方式实现：
1，在第一阶段，首先zookeeper的客户端和这个服务端的leader主结点会通过nio或者netty的方式建立连接，然后客户端可以向这个主结点里面发送数据。
2，主结点接收到数据之后，leader主结点会向这个从结点发送一个proposal的一个命令，并且会以轮询的方式发给所有从结点，同时会将这个data数据和事务id一起发送给从结点
3，主结点发送完这个命令之后，leader主结点会同步将数据存在本地磁盘里面，并且给自己投一个ack的票
4，从结点将主结点发来的数据会先存储在本地磁盘，并且给主结点返回一个ack
5，主结点会去统计这个ack的票数，就是从结点所返回的ack和自己投票的ack
6，在第二阶段，如果ack的票数大于一半，那么主结点就会给从结点发送一个commit提交命令
7，在从结点里面的数据一开始是存储在磁盘的，在接收到这个commit命令之后，会将数据存储到内存
8，主结点也会将存储在磁盘的数据加入这个内存里面
9，主结点最后会给客户端一个数据变动的Event事件，并给这个客户端返回一个命令 *** 作的结果
2，zookeeper的脑分裂问题就是说在一段很短的时间内，这个网络不稳定或者说这个出现这个断网的现象，那么可能造成leader和follow无法通信的情况，那么的从结点就会认为这个主结点可能挂了，因此集群的从结点就会重新进行一个主结点的选举，在短时间内，这个之前的主结点又恢复了，那么此时会有两个这个主结点，就是造成了这个脑裂问题，这样就会有大量的数据丢失。
解决答案 ：就是通过这个zab解决。就是说在如果出现脑分裂，那么就会有两个主结点，其中后面这个新选举的主结点会有从结点，而这个出现网络故障的主结点没有这个从结点。根据这个两阶段提交，如果外面有数据写进来，会先写到磁盘里面，此时两个主结点磁盘都有数据，但是需要通过投票机制，超过一半的投票才能将数据提交到内存里面，这个没有从结点的主结点获取的ack票数不能超过一半，那么就不能触发commit提交，那么就不能将数据加载到内存，就不会出现这个数据丢失的情况。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）