当前位置：首页 > news >正文

org.apache.zookeeper.server.quorum.QuorumPeerMain

news 文章来源：https://blog.csdn.net/hay23455/article/details/144763975 2025/5/26 1:10:09

QuorumPeerMain源代码

package org.apache.zookeeper.server.quorum;import java.io.IOException;
import javax.management.JMException;
import javax.security.sasl.SaslException;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.zookeeper.audit.ZKAuditProvider;
import org.apache.zookeeper.jmx.ManagedUtil;
import org.apache.zookeeper.metrics.MetricsProvider;
import org.apache.zookeeper.metrics.MetricsProviderLifeCycleException;
import org.apache.zookeeper.metrics.impl.MetricsProviderBootstrap;
import org.apache.zookeeper.server.DatadirCleanupManager;
import org.apache.zookeeper.server.ExitCode;
import org.apache.zookeeper.server.ServerCnxnFactory;
import org.apache.zookeeper.server.ServerMetrics;
import org.apache.zookeeper.server.ZKDatabase;
import org.apache.zookeeper.server.ZooKeeperServerMain;
import org.apache.zookeeper.server.admin.AdminServer.AdminServerException;
import org.apache.zookeeper.server.auth.ProviderRegistry;
import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
import org.apache.zookeeper.server.persistence.FileTxnSnapLog.DatadirException;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
import org.apache.zookeeper.server.util.JvmPauseMonitor;
import org.apache.zookeeper.util.ServiceUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/**** <h2>Configuration file</h2>** When the main() method of this class is used to start the program, the first* argument is used as a path to the config file, which will be used to obtain* configuration information. This file is a Properties file, so keys and* values are separated by equals (=) and the key/value pairs are separated* by new lines. The following is a general summary of keys used in the* configuration file. For full details on this see the documentation in* docs/index.html* <ol>* <li>dataDir - The directory where the ZooKeeper data is stored.</li>* <li>dataLogDir - The directory where the ZooKeeper transaction log is stored.</li>* <li>clientPort - The port used to communicate with clients.</li>* <li>tickTime - The duration of a tick in milliseconds. This is the basic* unit of time in ZooKeeper.</li>* <li>initLimit - The maximum number of ticks that a follower will wait to* initially synchronize with a leader.</li>* <li>syncLimit - The maximum number of ticks that a follower will wait for a* message (including heartbeats) from the leader.</li>* <li>server.<i>id</i> - This is the host:port[:port] that the server with the* given id will use for the quorum protocol.</li>* </ol>* In addition to the config file. There is a file in the data directory called* "myid" that contains the server id as an ASCII decimal value.**/
@InterfaceAudience.Public
public class QuorumPeerMain {private static final Logger LOG = LoggerFactory.getLogger(QuorumPeerMain.class);private static final String USAGE = "Usage: QuorumPeerMain configfile";protected QuorumPeer quorumPeer;/*** 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);ZKAuditProvider.addServerStartFailureAuditLog();ServiceUtils.requestSystemExit(ExitCode.INVALID_INVOCATION.getValue());} catch (ConfigException e) {LOG.error("Invalid config, exiting abnormally", e);System.err.println("Invalid config, exiting abnormally");ZKAuditProvider.addServerStartFailureAuditLog();ServiceUtils.requestSystemExit(ExitCode.INVALID_INVOCATION.getValue());} catch (DatadirException e) {LOG.error("Unable to access datadir, exiting abnormally", e);System.err.println("Unable to access datadir, exiting abnormally");ZKAuditProvider.addServerStartFailureAuditLog();ServiceUtils.requestSystemExit(ExitCode.UNABLE_TO_ACCESS_DATADIR.getValue());} catch (AdminServerException e) {LOG.error("Unable to start AdminServer, exiting abnormally", e);System.err.println("Unable to start AdminServer, exiting abnormally");ZKAuditProvider.addServerStartFailureAuditLog();ServiceUtils.requestSystemExit(ExitCode.ERROR_STARTING_ADMIN_SERVER.getValue());} catch (Exception e) {LOG.error("Unexpected exception, exiting abnormally", e);ZKAuditProvider.addServerStartFailureAuditLog();ServiceUtils.requestSystemExit(ExitCode.UNEXPECTED_ERROR.getValue());}LOG.info("Exiting normally");ServiceUtils.requestSystemExit(ExitCode.EXECUTION_FINISHED.getValue());}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 taskDatadirCleanupManager 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 standaloneZooKeeperServerMain.main(args);}}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, myid=" + config.getServerId());final MetricsProvider metricsProvider;try {metricsProvider = MetricsProviderBootstrap.startMetricsProvider(config.getMetricsProviderClassName(),config.getMetricsProviderConfiguration());} catch (MetricsProviderLifeCycleException error) {throw new IOException("Cannot boot MetricsProvider " + config.getMetricsProviderClassName(), error);}try {ServerMetrics.metricsProviderInitialized(metricsProvider);ProviderRegistry.initialize();ServerCnxnFactory cnxnFactory = null;ServerCnxnFactory secureCnxnFactory = null;if (config.getClientPortAddress() != null) {cnxnFactory = ServerCnxnFactory.createFactory();cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns(), config.getClientPortListenBacklog(), false);}if (config.getSecureClientPortAddress() != null) {secureCnxnFactory = ServerCnxnFactory.createFactory();secureCnxnFactory.configure(config.getSecureClientPortAddress(), config.getMaxClientCnxns(), config.getClientPortListenBacklog(), true);}quorumPeer = getQuorumPeer();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.setConnectToLearnerMasterLimit(config.getConnectToLearnerMasterLimit());quorumPeer.setObserverMasterPort(config.getObserverMasterPort());quorumPeer.setConfigFileName(config.getConfigFilename());quorumPeer.setClientPortListenBacklog(config.getClientPortListenBacklog());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.setSslQuorum(config.isSslQuorum());quorumPeer.setUsePortUnification(config.shouldUsePortUnification());quorumPeer.setLearnerType(config.getPeerType());quorumPeer.setSyncEnabled(config.getSyncEnabled());quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());if (config.sslQuorumReloadCertFiles) {quorumPeer.getX509Util().enableCertFileReloading();}quorumPeer.setMultiAddressEnabled(config.isMultiAddressEnabled());quorumPeer.setMultiAddressReachabilityCheckEnabled(config.isMultiAddressReachabilityCheckEnabled());quorumPeer.setMultiAddressReachabilityCheckTimeoutMs(config.getMultiAddressReachabilityCheckTimeoutMs());// sets quorum sasl authentication configurationsquorumPeer.setQuorumSaslEnabled(config.quorumEnableSasl);if (quorumPeer.isQuorumSaslAuthEnabled()) {quorumPeer.setQuorumServerSaslRequired(config.quorumServerRequireSasl);quorumPeer.setQuorumLearnerSaslRequired(config.quorumLearnerRequireSasl);quorumPeer.setQuorumServicePrincipal(config.quorumServicePrincipal);quorumPeer.setQuorumServerLoginContext(config.quorumServerLoginContext);quorumPeer.setQuorumLearnerLoginContext(config.quorumLearnerLoginContext);}org.apache.zookeeper.server.quorum.QuorumPeerMain.setQuorumCnxnThreadsSize(config.quorumCnxnThreadsSize);quorumPeer.initialize();if (config.jvmPauseMonitorToRun) {quorumPeer.setJvmPauseMonitor(new JvmPauseMonitor(config));}quorumPeer.start();ZKAuditProvider.addZKStartStopAuditLog();quorumPeer.join();} catch (InterruptedException e) {// warn, but generally this is okLOG.warn("Quorum Peer interrupted", e);} finally {try {metricsProvider.stop();} catch (Throwable error) {LOG.warn("Error while stopping metrics", error);}}}// @VisibleForTestingprotected QuorumPeer getQuorumPeer() throws SaslException {return new QuorumPeer();}/*** Shutdowns properly the service, this method is not a public API.*/public void close() {if (quorumPeer != null) {try {quorumPeer.shutdown();} finally {quorumPeer = null;}}}@Overridepublic String toString() {QuorumPeer peer = quorumPeer;return peer == null ? "" : peer.toString();}}

org.apache.zookeeper.server.quorum.QuorumPeerMain 是 Apache ZooKeeper 项目中的一个类，它负责启动和管理 ZooKeeper 的 Quorum 服务器。

ZooKeeper 是一个为分布式系统提供服务的开源项目，它为分布式应用提供了一致的、可靠的、高性能的协调服务。Quorum 服务器是 ZooKeeper 集群的核心组件之一，它负责处理客户端的请求，维护集群的状态，以及在集群中复制和同步数据。

QuorumPeerMain 类是 ZooKeeper Quorum 服务器的入口点。当你在命令行中运行 ZooKeeper 时，这个类会被 JVM 加载并执行。它负责解析命令行参数，配置 ZooKeeper 服务器，然后启动服务器并进入监听状态。

这个类包含了启动和管理 ZooKeeper Quorum 服务器的逻辑。它读取配置文件（通常是 zoo.cfg），设置必要的系统属性，创建和配置 QuorumPeer 对象，然后启动服务器。一旦服务器启动并运行，它将开始接受客户端的连接请求并处理这些请求。