Zookeeper源码学习

源码下载：https://archive.apache.org/dist/zookeeper/zookeeper-3.5.7/apache-zookeeper-3.5.7.tar.gz

1、持久化源码：

Leader和Follower中的数据会在内存和磁盘中各保存一份。所以需要将内存中的数据持久化到磁盘中。

在zookeeper-server\org.apache.zookeeper.server.persistence包下的相关类都是序列化相关的代码。

1）快照：

package org.apache.zookeeper.server.persistence;
import java.io.File;
import java.io.IOException;
import java.util.Map;

import org.apache.zookeeper.server.DataTree;

public interface SnapShot {
    
    /**
     * 反序列化
     */
    long deserialize(DataTree dt, Map<Long, Integer> sessions) 
        throws IOException;
    
    /**
     * 序列化
     */
    void serialize(DataTree dt, Map<Long, Integer> sessions, 
            File name) 
        throws IOException;
    
    /**
     * 查找最近的快照文件
     */
    File findMostRecentSnapshot() throws IOException;
    
    /**
     *释放资源
     */
    void close() throws IOException;
} 

2）操作日志：

package org.apache.zookeeper.server.persistence;

import java.io.IOException;

import org.apache.jute.Record;
import org.apache.zookeeper.server.ServerStats;
import org.apache.zookeeper.txn.TxnHeader;


public interface TxnLog {

    /**
     * 设置服务状态
     */
    void setServerStats(ServerStats serverStats);
    
    /**
     * 滚动日志
     */
    void rollLog() throws IOException;

    /**
     * 追加日志
     */
    boolean append(TxnHeader hdr, Record r) throws IOException;

    /**
     * 读取数据
     */
    TxnIterator read(long zxid) throws IOException;
    
    /**
     * 获取事务ID
     */
    long getLastLoggedZxid() throws IOException;
    
    /**
     * 删除事务
     */
    boolean truncate(long zxid) throws IOException;
    
    /**
     * 获取事务日志的数据库ID
     */
    long getDbId() throws IOException;
    
    /**
     * 提交
     */
    void commit() throws IOException;

    /**
     * 获取事务日志的过期同步时间(毫秒为单位)
     */
    long getTxnLogSyncElapsedTime();
   
    /** 
     * 关闭事务日志
     */
    void close() throws IOException;

    /**
     * 读取事务日志的迭代器接口 
     */
    public interface TxnIterator {
        /**
         * 获取事务头部
         */
        TxnHeader getHeader();
        
        /**
         * 获取事务
         */
        Record getTxn();
     
        /**
         * 下个事务
         */
        boolean next() throws IOException;
        
        /**
         * 关闭文件释放资源
         */
        void close() throws IOException;
        
        /**
         * Get an estimated storage space used to store transaction records
         * that will return by this iterator
         * @throws IOException
         */
        long getStorageSize() throws IOException;
    }
}

2、序列化源码

zookeeper-jute代码是关于Zookeeper序列化相关的源码

1）序列化和反序列化

package org.apache.jute;

import org.apache.yetus.audience.InterfaceAudience;

import java.io.IOException;


@InterfaceAudience.Public
public interface Record {
    //序列化
    public void serialize(OutputArchive archive, String tag)
        throws IOException;

    //反序列化
    public void deserialize(InputArchive archive, String tag)
        throws IOException;
}

3、Zookeeper服务端初始化源码

4、Zookeeper选举源码

 Zookeeper选举准备阶段

 Zookeeper选举执行阶段

 5、Follower和Leader状态同步源码

        当选举结束后，每个节点都需要根据自己的角色更新自己的状态。选举出的Leader更新自己的状态为Leader，其它节点更新自己状态为Follower。

        Leader更新状态入口：leader.lead()

        Follower更新状态入口：follower.followerLeader()

        注意：

        （1）Follower必须让Leader知道自己的状态：epoch(任期)、zxid、sid

        必须要找出谁是Leader；

        发起请求连接Leader；

        发送自己的信息给Leader；

        Leader接收到信息，必须要返回对应的信息给Follower。

         （2）当Leader第二种Follower的状态了，就确定需要做何种方式的数据同步DIFF、TRUNC、SNAP

         （3）执行数据同步

         （4）当Leader接收到超过半数Follower的ack之后，进入正常工作状态，集群启动完成了。

        最终总结同步的方式：

          （1）DIFF咱俩一样，不需要做什么

          （2）TRUNC Follower的zxid比Leader的zxid大，所以Follower要回滚

          （3） COMMIT Leader的zxid比Follower的zxid大，发送Proposal给Follower提交执行    

          （4）如果Follower并没有任何数据，直接使用SNAP的方式来执行数据同步（直接把数据全部序列到Follower）

6、服务端Leader启动

 7、服务端Follower启动

 8、客户端初始化源码