您当前的位置:首页 > 电脑百科 > 程序开发 > 语言 > JAVA

zookeeper源码分析之一服务端启动过程

时间:2019-09-03 14:06:46  来源:  作者:

zookeeper简介

zookeeper是为分布式应用提供分布式协作服务的开源软件。它提供了一组简单的原子操作,分布式应用可以基于这些原子操作来实现更高层次的同步服务,配置维护,组管理和命名。zookeeper的设计使基于它的编程非常容易,若我们熟悉目录树结构的文件系统,也会很容易使用zookeeper的数据模型样式。它运行在JAVA上,有java和c的客户端。

协作服务因难于获取正确而臭名远扬,他们特别易于出错如竞争条件和死锁。zookeeper的动机是减轻分布式应用中从零开始实现协作服务的压力。

zookeeper的特点

1.简单:zookeeper运行分布式进行通过一个共享的层次命名空间来进行协作,该命名空间的组织类似于标准的文件系统。命名空间包括数据注册器(称之为znode),在zookeeper看来,这类似于文件和目录。与典型的文件系统设计用来存储不同的是,zookeeper数据是存放在内存中,这意味着zookeeper可以实现很高的吞吐量和低延迟。

ZooKeeper 实现在高性能,高可用性,严格有序的访问方面有很大的优势。在性能方面的优势使它可以应用在大型的的分布式系统。在可靠性方面,避免单点故障。严格的顺序访问使它在客户端可以实现复杂的同步原语。

2. 可复制:类似于分布式进程的协作,zookeeper本身很容易在一组主机(称之为集合)中实现复制。zookeeper服务示意图:

zookeeper源码分析之一服务端启动过程

 

组成ZooKeeper服务的一组服务器都必须知道对方的。它们保存了内存映像的状态,以及在持久存储中的事务日志和快照。只要大部分的服务器可用,ZooKeeper服务将可用。

客户端连接到一台ZooKeeper服务器。客户端维护一个TCP连接,通过它发送请求,得到响应,得到监视事件,并发送心跳。如果TCP连接到服务器中断,客户端可以连接到不同的服务器。

3. 有序:ZooKeeper给每次更新使用数字打标记,它反映了所有zookeeper事务的顺序。随后的操作可以使用这些顺序来实现更高级别的抽象,如同步原语。

4.快速:它特别快,在“读为主”的工作中,ZooKeeper 应用程序运行在数千台机器,它在读远比写更多的时候(在10:1的比例)表现的最好。

数据模型与层次命名空间

ZooKeeper提供的名称空间更像是一个标准的文件系统。一个名字是一个由一个(或)分隔的路径元素的序列。zookeeper名称空间的每个节点由路径来标示。

zookeeper源码分析之一服务端启动过程

 

节点和临时节点

不像标准的文件系统,在ZooKeeper 命名空间中每个节点都有与它相关的数据以及子节点。它就像这样一个文件系统,它允许一个文件也可以是一个目录。(zookeeper是用来储存协作数据:状态信息,配置,位置信息等,因此,存储在每个节点的数据通常是很小的,在字节到千字节范围。)我们使用术语znode来表明我们谈论的是zookeeper数据节点。

znodes保存一个数据结构,该数据结构包括数据变化的版本号和时间戳,ACL的变化,这些信息允许缓存验证和协作更新。一个znode的数据的每次变化,版本号的增加。例如,每当客户检索数据时,它也接收到数据的版本。

在一个命名空间中的每个节点存储的数据的读写都是原子性的。读获取一个Znode所有的数据字节;写替换所有的数据。每个节点都有一个访问控制列表(ACL),限制谁可以做什么。

zookeeper也有临时节点的概念。这些znodes只要创建znode的会话是活跃的,它就存在的。当会话结束时,这些znode被删除。

条件更新与监控

ZooKeeper支持监控的概念。客户端可以在一个znode上设置一个监控。当znode发生变化时会触发或者移除监控。当监控触发时,客户端接收到一个报文,表明znode发生了变化。若客户端和一个zookeeper服务器的连接损坏时,客户端接收到一个本地通知。

保障

ZooKeeper非常快速和简单. 虽然它的目标是为建设更为复杂的服务,例如同步,它提供了一系列的保证。这些是:

  • 顺序一致性----客户端的更新将被应用于它们被发送的命令中。
  • 原子性-- - 更新要么成功要么失败,不存在部分成功或者部分失败.
  • 单系统映像 ---- 不管连接到哪台服务器,客户端看到相同的服务视图.
  • 可靠性---- 一旦一个更新发生,直到下次一个客户端重新了更新,否则从更新的时间后都会保持。
  • 及时性--- - 在一定时间范围内保证系统的客户视图是最新的.

简单api

zookeeper设计目标之一是提供一个简单的编程接口,因此,它只支持下面这些操作:

create

在节点树上某个位置上创建一个新的节点。

delete

删除一个节点

exists

测试某位置的节点是否存在

get data

从一个节点读取数据

set data

向一个节点写入数据

get children

检索一个节点的一组子节点

sync

等待数据传播至一致。

实现

zookeeper组件显示了zookeeper服务的高级组件。除了request processor,组成zookeeper服务的每个服务器复制它的每个组件的copy。

zookeeper组件

zookeeper源码分析之一服务端启动过程

 

replicated database是一个包含整个数据数的内存数据库. 为了可复原,更新被写到磁盘上,写操作在应用到内存数据库之前,先序列化到磁盘。

每个zookeeper服务器给所有的客户端提供服务。客户端恰恰连接到一个服务器来提交请求。读请求由每个服务器数据库的本地复制提供服务。写请求改变了服务的状态,由request processor来处理。

作为通信协议的一部分,所有客户端的写请求由一个单独的服务器处理,这个服务器是zookeeper的leader服务器,其余的zookeeper服务器叫做follower,follower从leader接收消息并达成消息传输。消息层在失败后替换leader并同步到连接到leader所有的follower。

ZooKeeper使用自定义的原子消息协议. 因消息层是原子性的, ZooKeeper 可以保证本地复制不会冲突. 当leader接收到一个写请求,当写操作应用到系统时,leader计算出系统的状态,并转化成一个捕捉新状态的事务.

zookeeper启动

服务端启动

bin/zkServer.sh start

其中,启动命令如下:

start)
 echo -n "Starting zookeeper ... "
 if [ -f "$ZOOPIDFILE" ]; then
 if kill -0 `cat "$ZOOPIDFILE"` > /dev/null 2>&1; then
 echo $command already running as process `cat "$ZOOPIDFILE"`.
 exit 0
 fi
 fi
 nohup "$JAVA" $ZOO_DATADIR_AUTOCREATE "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" 
 "-Dzookeeper.log.file=${ZOO_LOG_FILE}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" 
 -XX:+HeapDumpOnOutOfMemoryError -XX:OnOutOfMemoryError='kill -9 %p' 
 -cp "$CLASSPATH" $JVMFLAGS $ZOOMAIN "$ZOOCFG" > "$_ZOO_DAEMON_OUT" 2>&1 < /dev/null &
 if [ $? -eq 0 ]
 then
 if /bin/echo -n $! > "$ZOOPIDFILE"
 then
 sleep 1
 pid=$(cat "${ZOOPIDFILE}")
 if ps -p "${pid}" > /dev/null 2>&1; then
 echo STARTED
 else
 echo FAILED TO START
 exit 1
 fi
 else
 echo FAILED TO WRITE PID
 exit 1
 fi
 else
 echo SERVER DID NOT START
 exit 1
 fi
 ;;

其中:

ZOOMAIN 是启动程序的入口,其类为:

org.Apache.zookeeper.server.quorum.QuorumPeerMain

它的启动方法为:

 /**
 * To start the replicated server specify the configuration file name on
 * the command line.
 * @param args path to the configfile
 */
 public static void main(String[] args) {
 QuorumPeerMain main = new QuorumPeerMain();
 try {
 main.initializeAndRun(args);
 } catch (IllegalArgumentException e) {
 LOG.error("Invalid arguments, exiting abnormally", e);
 LOG.info(USAGE);
 System.err.println(USAGE);
 System.exit(2);
 } catch (ConfigException e) {
 LOG.error("Invalid config, exiting abnormally", e);
 System.err.println("Invalid config, exiting abnormally");
 System.exit(2);
 } catch (DatadirException e) {
 LOG.error("Unable to access datadir, exiting abnormally", e);
 System.err.println("Unable to access datadir, exiting abnormally");
 System.exit(3);
 } catch (AdminServerException e) {
 LOG.error("Unable to start AdminServer, exiting abnormally", e);
 System.err.println("Unable to start AdminServer, exiting abnormally");
 System.exit(4);
 } catch (Exception e) {
 LOG.error("Unexpected exception, exiting abnormally", e);
 System.exit(1);
 }
 LOG.info("Exiting normally");
 System.exit(0);
 }

调用初始化方法及run方法:

 protected void initializeAndRun(String[] args)
 throws ConfigException, IOException, AdminServerException
 {
 QuorumPeerConfig config = new QuorumPeerConfig();
 if (args.length == 1) {
 config.parse(args[0]);
 }
 // Start and schedule the the purge task
 DatadirCleanupManager purgeMgr = new DatadirCleanupManager(config
 .getDataDir(), config.getDataLogDir(), config
 .getSnapRetainCount(), config.getPurgeInterval());
 purgeMgr.start();
 if (args.length == 1 && config.isDistributed()) {
 runFromConfig(config);
 } else {
 LOG.warn("Either no config or no quorum defined in config, running "
 + " in standalone mode");
 // there is only server in the quorum -- run as standalone
 ZooKeeperServerMain.main(args);
 }
 }

上述代码主要分3部分:

1. 解析配置文件,默认的配置文件为上一级目录

config/zookeeper.properties或者config/zookeeper.cfg
 /**
 * Parse a ZooKeeper configuration file
 * @param path the patch of the configuration file
 * @throws ConfigException error processing configuration
 */
 public void parse(String path) throws ConfigException {
 LOG.info("Reading configuration from: " + path);
 
 try {
 File configFile = (new VerifyingFileFactory.Builder(LOG)
 .warnForRelativePath()
 .failForNonExistingPath()
 .build()).create(path);
 
 Properties cfg = new Properties();
 FileInputStream in = new FileInputStream(configFile);
 try {
 cfg.load(in);
 configFileStr = path;
 } finally {
 in.close();
 }
 
 parseProperties(cfg);
 } catch (IOException e) {
 throw new ConfigException("Error processing " + path, e);
 } catch (IllegalArgumentException e) {
 throw new ConfigException("Error processing " + path, e);
 } 
 
 if (dynamicConfigFileStr!=null) {
 try { 
 Properties dynamicCfg = new Properties();
 FileInputStream inConfig = new FileInputStream(dynamicConfigFileStr);
 try {
 dynamicCfg.load(inConfig);
 if (dynamicCfg.getProperty("version") != null) {
 throw new ConfigException("dynamic file shouldn't have version inside");
 }
 String version = getVersionFromFilename(dynamicConfigFileStr);
 // If there isn't any version associated with the filename,
 // the default version is 0.
 if (version != null) {
 dynamicCfg.setProperty("version", version);
 }
 } finally {
 inConfig.close();
 }
 setupQuorumPeerConfig(dynamicCfg, false);
 } catch (IOException e) {
 throw new ConfigException("Error processing " + dynamicConfigFileStr, e);
 } catch (IllegalArgumentException e) {
 throw new ConfigException("Error processing " + dynamicConfigFileStr, e);
 } 
 File nextDynamicConfigFile = new File(configFileStr + nextDynamicConfigFileSuffix);
 if (nextDynamicConfigFile.exists()) {
 try { 
 Properties dynamicConfigNextCfg = new Properties();
 FileInputStream inConfigNext = new FileInputStream(nextDynamicConfigFile); 
 try {
 dynamicConfigNextCfg.load(inConfigNext);
 } finally {
 inConfigNext.close();
 }
 boolean isHierarchical = false;
 for (Entry<Object, Object> entry : dynamicConfigNextCfg.entrySet()) {
 String key = entry.getKey().toString().trim(); 
 if (key.startsWith("group") || key.startsWith("weight")) {
 isHierarchical = true;
 break;
 }
 }
 lastSeenQuorumVerifier = createQuorumVerifier(dynamicConfigNextCfg, isHierarchical);
 } catch (IOException e) {
 LOG.warn("NextQuorumVerifier is initiated to null");
 }
 }
 }
 }

2. 启动安排清除任务

 // Start and schedule the the purge task
 DatadirCleanupManager purgeMgr = new DatadirCleanupManager(config
 .getDataDir(), config.getDataLogDir(), config
 .getSnapRetainCount(), config.getPurgeInterval());
 purgeMgr.start();

调用start方法:

/**
 * Validates the purge configuration and schedules the purge task. Purge
 * task keeps the most recent <code>snapRetainCount</code> number of
 * snapshots and deletes the remaining for every <code>purgeInterval</code>
 * hour(s).
 * <p>
 * <code>purgeInterval</code> of <code>0</code> or
 * <code>negative integer</code> will not schedule the purge task.
 * </p>
 * 
 * @see PurgeTxnLog#purge(File, File, int)
 */
 public void start() {
 if (PurgeTaskStatus.STARTED == purgeTaskStatus) {
 LOG.warn("Purge task is already running.");
 return;
 }
 // Don't schedule the purge task with zero or negative purge interval.
 if (purgeInterval <= 0) {
 LOG.info("Purge task is not scheduled.");
 return;
 }
 timer = new Timer("PurgeTask", true);
 TimerTask task = new PurgeTask(dataLogDir, snapDir, snapRetainCount);
 timer.scheduleAtFixedRate(task, 0, TimeUnit.HOURS.toMillis(purgeInterval));
 purgeTaskStatus = PurgeTaskStatus.STARTED;
 }

从上面代码可以看到,清除工作启动了一个定时器timer,PurgeTask继承实现了TimeTask(一个可以被定时器安排执行一次或者多次的task),PurgeTask的实现如下:

 static class PurgeTask extends TimerTask {
 private File logsDir;
 private File snapsDir;
 private int snapRetainCount;
 public PurgeTask(File dataDir, File snapDir, int count) {
 logsDir = dataDir;
 snapsDir = snapDir;
 snapRetainCount = count;
 }
 @Override
 public void run() {
 LOG.info("Purge task started.");
 try {
 PurgeTxnLog.purge(logsDir, snapsDir, snapRetainCount);
 } catch (Exception e) {
 LOG.error("Error occured while purging.", e);
 }
 LOG.info("Purge task completed.");
 }
 }

调用purge方法:

/**
 * Purges the snapshot and logs keeping the last num snapshots and the
 * corresponding logs. If logs are rolling or a new snapshot is created
 * during this process, these newest N snapshots or any data logs will be
 * excluded from current purging cycle.
 *
 * @param dataDir the dir that has the logs
 * @param snapDir the dir that has the snapshots
 * @param num the number of snapshots to keep
 * @throws IOException
 */
 public static void purge(File dataDir, File snapDir, int num) throws IOException {
 if (num < 3) {
 throw new IllegalArgumentException(COUNT_ERR_MSG);
 }
 FileTxnSnapLog txnLog = new FileTxnSnapLog(dataDir, snapDir);
 List<File> snaps = txnLog.findNRecentSnapshots(num);
 retainNRecentSnapshots(txnLog, snaps);
 }

先获取日志文件和快照,然后调用retainNRecentSnapshots方法处理:

 static void retainNRecentSnapshots(FileTxnSnapLog txnLog, List<File> snaps) {
 // found any valid recent snapshots?
 if (snaps.size() == 0)
 return;
 File snapShot = snaps.get(snaps.size() -1);
 final long leastZxidToBeRetain = Util.getZxidFromName(
 snapShot.getName(), PREFIX_SNAPSHOT);
 class MyFileFilter implements FileFilter{
 private final String prefix;
 MyFileFilter(String prefix){
 this.prefix=prefix;
 }
 public boolean accept(File f){
 if(!f.getName().startsWith(prefix + "."))
 return false;
 long fZxid = Util.getZxidFromName(f.getName(), prefix);
 if (fZxid >= leastZxidToBeRetain) {
 return false;
 }
 return true;
 }
 }
 // add all non-excluded log files
 List<File> files = new ArrayList<File>(Arrays.asList(txnLog
 .getDataDir().listFiles(new MyFileFilter(PREFIX_LOG))));
 // add all non-excluded snapshot files to the deletion list
 files.addAll(Arrays.asList(txnLog.getSnapDir().listFiles(
 new MyFileFilter(PREFIX_SNAPSHOT))));
 // remove the old files
 for(File f: files)
 {
 System.out.println("Removing file: "+
 DateFormat.getDateTimeInstance().format(f.lastModified())+
 "t"+f.getPath());
 if(!f.delete()){
 System.err.println("Failed to remove "+f.getPath());
 }
 }
 }

3. 启动zookeeper 服务器

3.1 启动单机

 /*
 * Start up the ZooKeeper server.
 *
 * @param args the configfile or the port datadir [ticktime]
 */
 public static void main(String[] args) {
 ZooKeeperServerMain main = new ZooKeeperServerMain();
 try {
 main.initializeAndRun(args);
 } catch (IllegalArgumentException e) {
 LOG.error("Invalid arguments, exiting abnormally", e);
 LOG.info(USAGE);
 System.err.println(USAGE);
 System.exit(2);
 } catch (ConfigException e) {
 LOG.error("Invalid config, exiting abnormally", e);
 System.err.println("Invalid config, exiting abnormally");
 System.exit(2);
 } catch (DatadirException e) {
 LOG.error("Unable to access datadir, exiting abnormally", e);
 System.err.println("Unable to access datadir, exiting abnormally");
 System.exit(3);
 } catch (AdminServerException e) {
 LOG.error("Unable to start AdminServer, exiting abnormally", e);
 System.err.println("Unable to start AdminServer, exiting abnormally");
 System.exit(4);
 } catch (Exception e) {
 LOG.error("Unexpected exception, exiting abnormally", e);
 System.exit(1);
 }
 LOG.info("Exiting normally");
 System.exit(0);
 }

调用方法:

 protected void initializeAndRun(String[] args)
 throws ConfigException, IOException, AdminServerException
 {
 try {
 ManagedUtil.registerLog4jMBeans();
 } catch (JMException e) {
 LOG.warn("Unable to register log4j JMX control", e);
 }
 ServerConfig config = new ServerConfig();
 if (args.length == 1) {
 config.parse(args[0]);
 } else {
 config.parse(args);
 }
 runFromConfig(config);
 }

启动过程:

 /**
 * Run from a ServerConfig.
 * @param config ServerConfig to use.
 * @throws IOException
 * @throws AdminServerException
 */
 public void runFromConfig(ServerConfig config) throws IOException, AdminServerException {
 LOG.info("Starting server");
 FileTxnSnapLog txnLog = null;
 try {
 // Note that this thread isn't going to be doing anything else,
 // so rather than spawning another thread, we will just call
 // run() in this thread.
 // create a file logger url from the command line args
 txnLog = new FileTxnSnapLog(config.dataLogDir, config.dataDir);
 ZooKeeperServer zkServer = new ZooKeeperServer( txnLog,
 config.tickTime, config.minSessionTimeout, config.maxSessionTimeout, null);
 // Start Admin server
 adminServer = AdminServerFactory.createAdminServer();
 adminServer.setZooKeeperServer(zkServer);
 adminServer.start();
 boolean needStartZKServer = true;
 if (config.getClientPortAddress() != null) {
 cnxnFactory = ServerCnxnFactory.createFactory();
 cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns(), false);
 cnxnFactory.startup(zkServer);
 // zkServer has been started. So we don't need to start it again in secureCnxnFactory.
 needStartZKServer = false;
 }
 if (config.getSecureClientPortAddress() != null) {
 secureCnxnFactory = ServerCnxnFactory.createFactory();
 secureCnxnFactory.configure(config.getSecureClientPortAddress(), config.getMaxClientCnxns(), true);
 secureCnxnFactory.startup(zkServer, needStartZKServer);
 }
 containerManager = new ContainerManager(zkServer.getZKDatabase(), zkServer.firstProcessor,
 Integer.getInteger("znode.container.checkIntervalMs", (int) TimeUnit.MINUTES.toMillis(1)),
 Integer.getInteger("znode.container.maxPerMinute", 10000)
 );
 containerManager.start();
 if (cnxnFactory != null) {
 cnxnFactory.join();
 }
 if (secureCnxnFactory != null) {
 secureCnxnFactory.join();
 }
 if (zkServer.isRunning()) {
 zkServer.shutdown();
 }
 } catch (InterruptedException e) {
 // warn, but generally this is ok
 LOG.warn("Server interrupted", e);
 } finally {
 if (txnLog != null) {
 txnLog.close();
 }
 }
 }
cnxnFactory.startup(zkServer);[NettyServerCnxnFactory]
 @Override
 public void startup(ZooKeeperServer zks, boolean startServer)
 throws IOException, InterruptedException {
 start();
 setZooKeeperServer(zks);
 if (startServer) {
 zks.startdata();
 zks.startup();
 }
 }
 public synchronized void startup() {
 if (sessionTracker == null) {
 createSessionTracker();
 }
 startSessionTracker();
 setupRequestProcessors();
 registerJMX();
 state = State.RUNNING;
 notifyAll();
 }
 protected void setupRequestProcessors() {
 RequestProcessor finalProcessor = new FinalRequestProcessor(this);
 RequestProcessor syncProcessor = new SyncRequestProcessor(this,
 finalProcessor);
 ((SyncRequestProcessor)syncProcessor).start();
 firstProcessor = new PrepRequestProcessor(this, syncProcessor);
 ((PrepRequestProcessor)firstProcessor).start();
 }

3.2 集群启动

 public void runFromConfig(QuorumPeerConfig config) throws IOException, AdminServerException {
 try {
 ManagedUtil.registerLog4jMBeans();
 } catch (JMException e) {
 LOG.warn("Unable to register log4j JMX control", e);
 }
 LOG.info("Starting quorum peer");
 try {
 ServerCnxnFactory cnxnFactory = null;
 ServerCnxnFactory secureCnxnFactory = null;
 if (config.getClientPortAddress() != null) {
 cnxnFactory = ServerCnxnFactory.createFactory();
 cnxnFactory.configure(config.getClientPortAddress(),
 config.getMaxClientCnxns(),
 false);
 }
 if (config.getSecureClientPortAddress() != null) {
 secureCnxnFactory = ServerCnxnFactory.createFactory();
 secureCnxnFactory.configure(config.getSecureClientPortAddress(),
 config.getMaxClientCnxns(),
 true);
 }
 quorumPeer = new QuorumPeer();
 quorumPeer.setTxnFactory(new FileTxnSnapLog(
 config.getDataLogDir(),
 config.getDataDir()));
 quorumPeer.enableLocalSessions(config.areLocalSessionsEnabled());
 quorumPeer.enableLocalSessionsUpgrading(
 config.isLocalSessionsUpgradingEnabled());
 //quorumPeer.setQuorumPeers(config.getAllMembers());
 quorumPeer.setElectionType(config.getElectionAlg());
 quorumPeer.setMyid(config.getServerId());
 quorumPeer.setTickTime(config.getTickTime());
 quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
 quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
 quorumPeer.setInitLimit(config.getInitLimit());
 quorumPeer.setSyncLimit(config.getSyncLimit());
 quorumPeer.setConfigFileName(config.getConfigFilename());
 quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
 quorumPeer.setQuorumVerifier(config.getQuorumVerifier(), false);
 if (config.getLastSeenQuorumVerifier()!=null) {
 quorumPeer.setLastSeenQuorumVerifier(config.getLastSeenQuorumVerifier(), false);
 }
 quorumPeer.initConfigInZKDatabase();
 quorumPeer.setCnxnFactory(cnxnFactory);
 quorumPeer.setSecureCnxnFactory(secureCnxnFactory);
 quorumPeer.setLearnerType(config.getPeerType());
 quorumPeer.setSyncEnabled(config.getSyncEnabled());
 quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());
 
 quorumPeer.start();
 quorumPeer.join();
 } catch (InterruptedException e) {
 // warn, but generally this is ok
 LOG.warn("Quorum Peer interrupted", e);
 }
 }

从上述代码可以看出,QuorumPeer的start()方法和join()方法是主流程。

QuorumPeer继承了ZooKeeperThread,ZooKeeperThread继承自Thread,故QuorumPeer间接继承了Thread。

 @Override
 public synchronized void start() {
 if (!getView().containsKey(myid)) {
 throw new RuntimeException("My id " + myid + " not in the peer list");
 }
 loadDataBase();
 startServerCnxnFactory();
 try {
 adminServer.start();
 } catch (AdminServerException e) {
 LOG.warn("Problem starting AdminServer", e);
 System.out.println(e);
 }
 startLeaderElection();
 super.start();
 }

3.2.1. 启动时先从内存数据库中恢复数据

 private void loadDataBase() {
 try {
 zkDb.loadDataBase();
 // load the epochs
 long lastProcessedZxid = zkDb.getDataTree().lastProcessedZxid;
 long epochOfZxid = ZxidUtils.getEpochFromZxid(lastProcessedZxid);
 try {
 currentEpoch = readLongFromFile(CURRENT_EPOCH_FILENAME);
 } catch(FileNotFoundException e) {
 // pick a reasonable epoch number
 // this should only hAppen once when moving to a
 // new code version
 currentEpoch = epochOfZxid;
 LOG.info(CURRENT_EPOCH_FILENAME
 + " not found! Creating with a reasonable default of {}. This should only happen when you are upgrading your installation",
 currentEpoch);
 writeLongToFile(CURRENT_EPOCH_FILENAME, currentEpoch);
 }
 if (epochOfZxid > currentEpoch) {
 throw new IOException("The current epoch, " + ZxidUtils.zxidToString(currentEpoch) + ", is older than the last zxid, " + lastProcessedZxid);
 }
 try {
 acceptedEpoch = readLongFromFile(ACCEPTED_EPOCH_FILENAME);
 } catch(FileNotFoundException e) {
 // pick a reasonable epoch number
 // this should only happen once when moving to a
 // new code version
 acceptedEpoch = epochOfZxid;
 LOG.info(ACCEPTED_EPOCH_FILENAME
 + " not found! Creating with a reasonable default of {}. This should only happen when you are upgrading your installation",
 acceptedEpoch);
 writeLongToFile(ACCEPTED_EPOCH_FILENAME, acceptedEpoch);
 }
 if (acceptedEpoch < currentEpoch) {
 throw new IOException("The accepted epoch, " + ZxidUtils.zxidToString(acceptedEpoch) + " is less than the current epoch, " + ZxidUtils.zxidToString(currentEpoch));
 }
 } catch(IOException ie) {
 LOG.error("Unable to load database on disk", ie);
 throw new RuntimeException("Unable to run quorum server ", ie);
 }
 }

调用

 /**
 * load the database from the disk onto memory and also add
 * the transactions to the committedlog in memory.
 * @return the last valid zxid on disk
 * @throws IOException
 */
 public long loadDataBase() throws IOException {
 PlayBackListener listener=new PlayBackListener(){
 public void onTxnLoaded(TxnHeader hdr,Record txn){
 Request r = new Request(0, hdr.getCxid(),hdr.getType(), hdr, txn, hdr.getZxid());
 addCommittedProposal(r);
 }
 };
 long zxid = snapLog.restore(dataTree,sessionsWithTimeouts,listener);
 initialized = true;
 return zxid;
 }
 /**
 * maintains a list of last <i>committedLog</i>
 * or so committed requests. This is used for
 * fast follower synchronization.
 * @param request committed request
 */
 public void addCommittedProposal(Request request) {
 WriteLock wl = logLock.writeLock();
 try {
 wl.lock();
 if (committedLog.size() > commitLogCount) {
 committedLog.removeFirst();
 minCommittedLog = committedLog.getFirst().packet.getZxid();
 }
 if (committedLog.isEmpty()) {
 minCommittedLog = request.zxid;
 maxCommittedLog = request.zxid;
 }
 ByteArrayOutputStream baos = new ByteArrayOutputStream();
 BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
 try {
 request.getHdr().serialize(boa, "hdr");
 if (request.getTxn() != null) {
 request.getTxn().serialize(boa, "txn");
 }
 baos.close();
 } catch (IOException e) {
 LOG.error("This really should be impossible", e);
 }
 QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid,
 baos.toByteArray(), null);
 Proposal p = new Proposal();
 p.packet = pp;
 p.request = request;
 committedLog.add(p);
 maxCommittedLog = p.packet.getZxid();
 } finally {
 wl.unlock();
 }
 }

3.2.2 启动NettyServerCnxnFactory绑定服务

 @Override
 public void start() {
 LOG.info("binding to port " + localAddress);
 parentChannel = bootstrap.bind(localAddress);
 }

3.2.3 选举算法

 synchronized public void startLeaderElection() {
 try {
 if (getPeerState() == ServerState.LOOKING) {
 currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
 }
 } catch(IOException e) {
 RuntimeException re = new RuntimeException(e.getMessage());
 re.setStackTrace(e.getStackTrace());
 throw re;
 }
 // if (!getView().containsKey(myid)) {
 // throw new RuntimeException("My id " + myid + " not in the peer list");
 //}
 if (electionType == 0) {
 try {
 udpSocket = new DatagramSocket(myQuorumAddr.getPort());
 responder = new ResponderThread();
 responder.start();
 } catch (SocketException e) {
 throw new RuntimeException(e);
 }
 }
 this.electionAlg = createElectionAlgorithm(electionType);
 }

调用

 @SuppressWarnings("deprecation")
 protected Election createElectionAlgorithm(int electionAlgorithm){
 Election le=null;
 //TODO: use a factory rather than a switch
 switch (electionAlgorithm) {
 case 0:
 le = new LeaderElection(this);
 break;
 case 1:
 le = new AuthFastLeaderElection(this);
 break;
 case 2:
 le = new AuthFastLeaderElection(this, true);
 break;
 case 3:
 qcm = new QuorumCnxManager(this);
 QuorumCnxManager.Listener listener = qcm.listener;
 if(listener != null){
 listener.start();
 FastLeaderElection fle = new FastLeaderElection(this, qcm);
 fle.start();
 le = fle;
 } else {
 LOG.error("Null listener when initializing cnx manager");
 }
 break;
 default:
 assert false;
 }
 return le;
 }

调用选举方法:

/**
 * Starts a new round of leader election. Whenever our QuorumPeer
 * changes its state to LOOKING, this method is invoked, and it
 * sends notifications to all other peers.
 */
 public Vote lookForLeader() throws InterruptedException {
 try {
 self.jmxLeaderElectionBean = new LeaderElectionBean();
 MBeanRegistry.getInstance().register(
 self.jmxLeaderElectionBean, self.jmxLocalPeerBean);
 } catch (Exception e) {
 LOG.warn("Failed to register with JMX", e);
 self.jmxLeaderElectionBean = null;
 }
 if (self.start_fle == 0) {
 self.start_fle = Time.currentElapsedTime();
 }
 try {
 HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();
 HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();
 int notTimeout = finalizeWait;
 synchronized(this){
 logicalclock.incrementAndGet();
 updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
 }
 LOG.info("New election. My id = " + self.getId() +
 ", proposed zxid=0x" + Long.toHexString(proposedZxid));
 sendNotifications();
 /*
 * Loop in which we exchange notifications until we find a leader
 */
 while ((self.getPeerState() == ServerState.LOOKING) &&
 (!stop)){
 /*
 * Remove next notification from queue, times out after 2 times
 * the termination time
 */
 Notification n = recvqueue.poll(notTimeout,
 TimeUnit.MILLISECONDS);
 /*
 * Sends more notifications if haven't received enough.
 * Otherwise processes new notification.
 */
 if(n == null){
 if(manager.haveDelivered()){
 sendNotifications();
 } else {
 manager.connectAll();
 }
 /*
 * Exponential backoff
 */
 int tmpTimeOut = notTimeout*2;
 notTimeout = (tmpTimeOut < maxNotificationInterval?
 tmpTimeOut : maxNotificationInterval);
 LOG.info("Notification time out: " + notTimeout);
 } 
 else if (self.getCurrentAndNextConfigVoters().contains(n.sid)) {
 /*
 * Only proceed if the vote comes from a replica in the current or next
 * voting view.
 */
 switch (n.state) {
 case LOOKING:
 // If notification > current, replace and send messages out
 if (n.electionEpoch > logicalclock.get()) {
 logicalclock.set(n.electionEpoch);
 recvset.clear();
 if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
 getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
 updateProposal(n.leader, n.zxid, n.peerEpoch);
 } else {
 updateProposal(getInitId(),
 getInitLastLoggedZxid(),
 getPeerEpoch());
 }
 sendNotifications();
 } else if (n.electionEpoch < logicalclock.get()) {
 if(LOG.isDebugEnabled()){
 LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
 + Long.toHexString(n.electionEpoch)
 + ", logicalclock=0x" + Long.toHexString(logicalclock.get()));
 }
 break;
 } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
 proposedLeader, proposedZxid, proposedEpoch)) {
 updateProposal(n.leader, n.zxid, n.peerEpoch);
 sendNotifications();
 }
 if(LOG.isDebugEnabled()){
 LOG.debug("Adding vote: from=" + n.sid +
 ", proposed leader=" + n.leader +
 ", proposed zxid=0x" + Long.toHexString(n.zxid) +
 ", proposed election epoch=0x" + Long.toHexString(n.electionEpoch));
 }
 recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
 if (termPredicate(recvset,
 new Vote(proposedLeader, proposedZxid,
 logicalclock.get(), proposedEpoch))) {
 // Verify if there is any change in the proposed leader
 while((n = recvqueue.poll(finalizeWait,
 TimeUnit.MILLISECONDS)) != null){
 if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
 proposedLeader, proposedZxid, proposedEpoch)){
 recvqueue.put(n);
 break;
 }
 }
 /*
 * This predicate is true once we don't read any new
 * relevant message from the reception queue
 */
 if (n == null) {
 self.setPeerState((proposedLeader == self.getId()) ?
 ServerState.LEADING: learningState());
 Vote endVote = new Vote(proposedLeader,
 proposedZxid, proposedEpoch);
 leaveInstance(endVote);
 return endVote;
 }
 }
 break;
 case OBSERVING:
 LOG.debug("Notification from observer: " + n.sid);
 break;
 case FOLLOWING:
 case LEADING:
 /*
 * Consider all notifications from the same epoch
 * together.
 */
 if(n.electionEpoch == logicalclock.get()){
 recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
 if(termPredicate(recvset, new Vote(n.leader,
 n.zxid, n.electionEpoch, n.peerEpoch, n.state))
 && checkLeader(outofelection, n.leader, n.electionEpoch)) {
 self.setPeerState((n.leader == self.getId()) ?
 ServerState.LEADING: learningState());
 Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
 leaveInstance(endVote);
 return endVote;
 }
 }
 /*
 * Before joining an established ensemble, verify that
 * a majority are following the same leader.
 * Only peer epoch is used to check that the votes come
 * from the same ensemble. This is because there is at
 * least one corner case in which the ensemble can be
 * created with inconsistent zxid and election epoch
 * info. However, given that only one ensemble can be
 * running at a single point in time and that each 
 * epoch is used only once, using only the epoch to 
 * compare the votes is sufficient.
 * 
 * @see https://issues.apache.org/jira/browse/ZOOKEEPER-1732
 */
 outofelection.put(n.sid, new Vote(n.leader, 
 IGNOREVALUE, IGNOREVALUE, n.peerEpoch, n.state));
 if (termPredicate(outofelection, new Vote(n.leader,
 IGNOREVALUE, IGNOREVALUE, n.peerEpoch, n.state))
 && checkLeader(outofelection, n.leader, IGNOREVALUE)) {
 synchronized(this){
 logicalclock.set(n.electionEpoch);
 self.setPeerState((n.leader == self.getId()) ?
 ServerState.LEADING: learningState());
 }
 Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
 leaveInstance(endVote);
 return endVote;
 }
 break;
 default:
 LOG.warn("Notification state unrecoginized: " + n.state
 + " (n.state), " + n.sid + " (n.sid)");
 break;
 }
 } else {
 LOG.warn("Ignoring notification from non-cluster member " + n.sid);
 }
 }
 return null;
 } finally {
 try {
 if(self.jmxLeaderElectionBean != null){
 MBeanRegistry.getInstance().unregister(
 self.jmxLeaderElectionBean);
 }
 } catch (Exception e) {
 LOG.warn("Failed to unregister with JMX", e);
 }
 self.jmxLeaderElectionBean = null;
 }
 }

4. 小结

本文先介绍了zookeeper开源分布式协作系统及其特点、应用场景,然后根据zookeeper的启动方式,找到zookeeper的入口。在入口方法中,单机启动使用ZooKeeperServerMain,最终调用ZookeeperServer的startup()方法来RequestProcessor;集群启动时调用QuorumPeer的start方法,接着也是调用ZookeeperServer的startup()方法来RequestProcessor,最后调用选举算法选出leader。

参考文献:

【1】http://zookeeper.apache.org/doc/r3.4.6/zookeeperOver.html

【2】http://zookeeper.apache.org/doc/r3.4.6/zookeeperStarted.html



Tags:zookeeper   点击:()  评论:()
声明:本站部分内容及图片来自互联网,转载是出于传递更多信息之目的,内容观点仅代表作者本人,如有任何标注错误或版权侵犯请与我们联系(Email:2595517585@qq.com),我们将及时更正、删除,谢谢。
▌相关推荐
一、准备三台机器这里我使用VirtualBox创建3个虚拟机来进行部署zk集群,VirtualBox不了解的可自行百度; 二、部署linux系统此处不讲解linux部署,很简单,百度一下很多教程的部署...【详细内容】
2021-12-08  Tags: zookeeper  点击:(17)  评论:(0)  加入收藏
zookeeper动物管理员,是一个很形象的名字,是一个分布式协调服务。它可以用来做分布式配置管理,服务注册及发现,分布式锁。在CAP中,属于CP型。下图是zookeeper的架构图: 图中,绿色的...【详细内容】
2021-11-16  Tags: zookeeper  点击:(38)  评论:(0)  加入收藏
环境:Spring Boot 2.3.9 + Spring Cloud Hoxton.SR8服务发现注册请参考《SpringCloud Zookeeper服务发现及负载均衡 》zookeeper安装配置请参考《Kafka(zookeeper)环境配置超级...【详细内容】
2021-04-06  Tags: zookeeper  点击:(276)  评论:(0)  加入收藏
ZK(zookeeper)是微服务解决方案中拥有服务注册发现最为核心的环境,是微服务的基石。作为服务注册发现模块,并不是只有ZK一种产品,目前得到行业认可的还有:Eureka、Consul。这里我...【详细内容】
2021-03-04  Tags: zookeeper  点击:(179)  评论:(0)  加入收藏
前三篇讲了Zookeeper的特性、客户端使用和集群原理、典型使用场景实践,本篇重点深入了解ZAB协议以及源码实现的解析。...【详细内容】
2020-10-08  Tags: zookeeper  点击:(85)  评论:(0)  加入收藏
某天程序员小白参加面试:几番苦战之后,面试进入白热化阶段。面试官大开大合,小白见招拆招。一时之间,难解难分,两人对拆数十回合不分胜负。说时迟,那时快,小白的左手像火焰一般炙热...【详细内容】
2020-08-18  Tags: zookeeper  点击:(112)  评论:(0)  加入收藏
一、zk是什么:1、个人理解zk=文件系统+通知机制。2、zk是一个分布式的应用程序协调服务,我理解的就是有两台集器A、B,A对一个数据进行了操作,B是如何知道的,这个就需要zk的支持。...【详细内容】
2020-08-11  Tags: zookeeper  点击:(58)  评论:(0)  加入收藏
典型应用场景Apache HBaseHBase是一个通常与Hadoop一起使用的数据存储仓库。在HBase中,ZooKeeper用于选举一个集群内的主节点,以便跟踪可用的服务器,并保存集群的元数据。Apach...【详细内容】
2020-07-29  Tags: zookeeper  点击:(45)  评论:(0)  加入收藏
如上图所示,kafaka集群的 broker,和 Consumer 都需要连接 Zookeeper。 Producer 直接连接 Broker。Producer 把数据上传到 Broker,Producer可以指定数据有几个分区、几个备份...【详细内容】
2020-06-15  Tags: zookeeper  点击:(125)  评论:(0)  加入收藏
本文主要分享一下zookeeper的一些基本概念,在正式进入正题前,和大家聊一聊刚入行时我的面试经验,可以说是耿直的有些可爱。面试官:用过zookeeper 吗?我:用过啊,给dubbo提供服务的...【详细内容】
2020-04-01  Tags: zookeeper  点击:(141)  评论:(0)  加入收藏
▌简易百科推荐
面向对象的特征之一封装 面向对象的特征之二继承 方法重写(override/overWrite) 方法的重载(overload)和重写(override)的区别: 面向对象特征之三:多态 Instanceof关键字...【详细内容】
2021-12-28  顶顶架构师    Tags:面向对象   点击:(2)  评论:(0)  加入收藏
一、Redis使用过程中一些小的注意点1、不要把Redis当成数据库来使用二、Arrays.asList常见失误需求:把数组转成list集合去处理。方法:Arrays.asList 或者 Java8的stream流式处...【详细内容】
2021-12-27  CF07    Tags:Java   点击:(3)  评论:(0)  加入收藏
文章目录 如何理解面向对象编程? JDK 和 JRE 有什么区别? 如何理解Java中封装,继承、多态特性? 如何理解Java中的字节码对象? 你是如何理解Java中的泛型的? 说说泛型应用...【详细内容】
2021-12-24  Java架构师之路    Tags:JAVA   点击:(5)  评论:(0)  加入收藏
大家好!我是老码农,一个喜欢技术、爱分享的同学,从今天开始和大家持续分享JVM调优方面的经验。JVM调优是个大话题,涉及的知识点很庞大 Java内存模型 垃圾回收机制 各种工具使用 ...【详细内容】
2021-12-23  小码匠和老码农    Tags:JVM调优   点击:(12)  评论:(0)  加入收藏
前言JDBC访问Postgresql的jsonb类型字段当然可以使用Postgresql jdbc驱动中提供的PGobject,但是这样在需要兼容多种数据库的系统开发中显得不那么通用,需要特殊处理。本文介绍...【详细内容】
2021-12-23  dingle    Tags:JDBC   点击:(13)  评论:(0)  加入收藏
Java与Lua相互调用案例比较少,因此项目使用需要做详细的性能测试,本内容只做粗略测试。目前已完成初版Lua-Java调用框架开发,后期有时间准备把框架进行抽象,并开源出来,感兴趣的...【详细内容】
2021-12-23  JAVA小白    Tags:Java   点击:(11)  评论:(0)  加入收藏
Java从版本5开始,在 java.util.concurrent.locks包内给我们提供了除了synchronized关键字以外的几个新的锁功能的实现,ReentrantLock就是其中的一个。但是这并不意味着我们可...【详细内容】
2021-12-17  小西学JAVA    Tags:JAVA并发   点击:(11)  评论:(0)  加入收藏
一、概述final是Java关键字中最常见之一,表示“最终的,不可更改”之意,在Java中也正是这个意思。有final修饰的内容,就会变得与众不同,它们会变成终极存在,其内容成为固定的存在。...【详细内容】
2021-12-15  唯一浩哥    Tags:Java基础   点击:(17)  评论:(0)  加入收藏
1、问题描述关于java中的日志管理logback,去年写过关于logback介绍的文章,这次项目中又优化了下,记录下,希望能帮到需要的朋友。2、解决方案这次其实是碰到了一个问题,一般的情况...【详细内容】
2021-12-15  软件老王    Tags:logback   点击:(19)  评论:(0)  加入收藏
本篇文章我们以AtomicInteger为例子,主要讲解下CAS(Compare And Swap)功能是如何在AtomicInteger中使用的,以及提供CAS功能的Unsafe对象。我们先从一个例子开始吧。假设现在我们...【详细内容】
2021-12-14  小西学JAVA    Tags:JAVA   点击:(22)  评论:(0)  加入收藏
最新更新
栏目热门
栏目头条