这篇文章主要介绍“zk中learner的作用是什么”，在日常操作中，相信很多人在zk中learner的作用是什么问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”zk中learner的作用是什么”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
创新互联是一家专业提供忠县企业网站建设,专注与网站设计、成都网站建设、成都h5网站建设、小程序制作等业务。10年已为忠县众多企业、政府机构等服务。创新互联专业的建站公司优惠进行中。
learner时observer,follower的父类，定义了公共属性和方法
子类 Follower 和Observer
内部类：
PacketInFlight表示在提议中还没有commit的消息
static class PacketInFlight { TxnHeader hdr; Record rec; }
属性：
QuorumPeer | 服务器节点 |
LearnerZooKeeperServer | learner的服务节点 |
BufferedOutputStream | 输出流 |
Socket | 端口套接字 |
InetSocketAddress | 地址信息 |
InputArchive | 输入存档 |
OutputArchive | 输出存档 |
leaderProtocolVersion | leader协议版本 |
BUFFERED_MESSAGE_SIZE | 缓存信息大小 |
MessageTracker | 顺序接收和发送信息 |
方法
validateSession(ServerCnxn cnxn, long clientId, int timeout) | 验证session有效性 |
writePacket(QuorumPacket pp, boolean flush) | 发送包给leader |
readPacket(QuorumPacket pp) | 从leader读取message |
request(Request request) | 发送request给leader |
findLeader | 查找认为是leader的地址信息 |
createSocket() | 创建socket对象 |
registerWithLeader(int pktType) | 执行handshake protocal建立follower/observer连接 |
到服务器验证session有效性
void validateSession(ServerCnxn cnxn, long clientId, int timeout) throws IOException {
LOG.info("Revalidating client: 0x" + Long.toHexString(clientId));
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(baos);
dos.writeLong(clientId);
dos.writeInt(timeout);
dos.close();
QuorumPacket qp = new QuorumPacket(Leader.REVALIDATE, -1, baos.toByteArray(), null);
pendingRevalidations.put(clientId, cnxn);
if (LOG.isTraceEnabled()) {
ZooTrace.logTraceMessage(
LOG,
ZooTrace.SESSION_TRACE_MASK,
"To validate session 0x" + Long.toHexString(clientId));
}
writePacket(qp, true);
}
void writePacket(QuorumPacket pp, boolean flush) throws IOException {
synchronized (leaderOs) {
if (pp != null) {
messageTracker.trackSent(pp.getType());
leaderOs.writeRecord(pp, "packet");
}
if (flush) {
bufferedOutput.flush();
}
}
}
void request(Request request) throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream oa = new DataOutputStream(baos);
oa.writeLong(request.sessionId);
oa.writeInt(request.cxid);
oa.writeInt(request.type);
if (request.request != null) {
request.request.rewind();
int len = request.request.remaining();
byte[] b = new byte[len];
request.request.get(b);
request.request.rewind();
oa.write(b);
}
oa.close();
QuorumPacket qp = new QuorumPacket(Leader.REQUEST, -1, baos.toByteArray(), request.authInfo);
writePacket(qp, true);
}
查找当前的leader信息
protected QuorumServer findLeader() {
QuorumServer leaderServer = null;
// Find the leader by id
Vote current = self.getCurrentVote();
for (QuorumServer s : self.getView().values()) {
if (s.id == current.getId()) {
// Ensure we have the leader's correct IP address before
// attempting to connect.
s.recreateSocketAddresses();
leaderServer = s;
break;
}
}
if (leaderServer == null) {
LOG.warn("Couldn't find the leader with id = " + current.getId());
}
return leaderServer;
}
连接套接字
sockConnect(Socket sock, InetSocketAddress addr, int timeout)
建立和leader的连接
/**
* Establish a connection with the LearnerMaster found by findLearnerMaster.
* Followers only connect to Leaders, Observers can connect to any active LearnerMaster.
* Retries until either initLimit time has elapsed or 5 tries have happened.
* @param addr - the address of the Peer to connect to.
* @throws IOException - if the socket connection fails on the 5th attempt
* if there is an authentication failure while connecting to leader
* @throws X509Exception
* @throws InterruptedException
*/
protected void connectToLeader(InetSocketAddress addr, String hostname) throws IOException, InterruptedException, X509Exception {
this.sock = createSocket();
this.leaderAddr = addr;
// leader connection timeout defaults to tickTime * initLimit
int connectTimeout = self.tickTime * self.initLimit;
// but if connectToLearnerMasterLimit is specified, use that value to calculate
// timeout instead of using the initLimit value
if (self.connectToLearnerMasterLimit > 0) {
connectTimeout = self.tickTime * self.connectToLearnerMasterLimit;
}
int remainingTimeout;
long startNanoTime = nanoTime();
for (int tries = 0; tries < 5; tries++) {
try {
// recalculate the init limit time because retries sleep for 1000 milliseconds
remainingTimeout = connectTimeout - (int) ((nanoTime() - startNanoTime) / 1000000);
if (remainingTimeout <= 0) {
LOG.error("connectToLeader exceeded on retries.");
throw new IOException("connectToLeader exceeded on retries.");
}
sockConnect(sock, addr, Math.min(connectTimeout, remainingTimeout));
if (self.isSslQuorum()) {
//开始握手
((SSLSocket) sock).startHandshake();
}
sock.setTcpNoDelay(nodelay);
break;
} catch (IOException e) {
//出现异常
remainingTimeout = connectTimeout - (int) ((nanoTime() - startNanoTime) / 1000000);
//剩余超时时间
if (remainingTimeout <= 1000) {
//打印错误日志
LOG.error("Unexpected exception, connectToLeader exceeded. tries=" + tries
+ ", remaining init limit=" + remainingTimeout
+ ", connecting to " + addr, e);
throw e;
//尝试次数大于4
} else if (tries >= 4) {
//打印错误日志
LOG.error("Unexpected exception, retries exceeded. tries=" + tries
+ ", remaining init limit=" + remainingTimeout
+ ", connecting to " + addr, e);
throw e;
} else {
//发出警告
LOG.warn("Unexpected exception, tries=" + tries
+ ", remaining init limit=" + remainingTimeout
+ ", connecting to " + addr, e);
//重新尝试建立socket连接
this.sock = createSocket();
}
}
//读取配置延时时间，默认100ns
Thread.sleep(leaderConnectDelayDuringRetryMs);
}
self.authLearner.authenticate(sock, hostname);
leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream(sock.getInputStream()));
bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
leaderOs = BinaryOutputArchive.getArchive(bufferedOutput);
}到此，关于“zk中learner的作用是什么”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注创新互联网站，小编会继续努力为大家带来更多实用的文章！
