BookKeeper（BK）启动流程

文章目录

BookKeeper（BK）启动流程
- 解析命令行参数
- 构建bookie所需的服务
- - 构建状态（指标）服务
  - 构建BookieService
  - - 构造内存分配器
    - 构造NettyServer
    - 构建Bookie

BK的启动入口类是Main，Main有一个静态代码块，在执行main方法之前，会限执行静态代码块的内容：

static final Options BK_OPTS = new Options();
static {BK_OPTS.addOption("c", "conf", true, "Configuration for Bookie Server");BK_OPTS.addOption("withAutoRecovery", false,"Start Autorecovery service Bookie server");BK_OPTS.addOption("r", "readOnly", false,"Force Start a ReadOnly Bookie server");BK_OPTS.addOption("z", "zkserver", true, "Zookeeper Server");BK_OPTS.addOption("m", "zkledgerpath", true, "Zookeeper ledgers root path");BK_OPTS.addOption("p", "bookieport", true, "bookie port exported");BK_OPTS.addOption("j", "journal", true, "bookie journal directory");Option indexDirs = new Option ("i", "indexdirs", true, "bookie index directories");indexDirs.setArgs(10);BK_OPTS.addOption(indexDirs);Option ledgerDirs = new Option ("l", "ledgerdirs", true, "bookie ledgers directories");ledgerDirs.setArgs(10);BK_OPTS.addOption(ledgerDirs);BK_OPTS.addOption("h", "help", false, "Print help message");}

这部分内容是构建出Options，供后续的启动参数解析使用，然后执行main()方法。

public static void main(String[] args) {int retCode = doMain(args);Runtime.getRuntime().exit(retCode);
}

核心启动逻辑都在doMain()方法里，

static int doMain(String[] args) {ServerConfiguration conf;// 0. parse command line， 解析命令行参数try {conf = parseCommandLine(args);} catch (IllegalArgumentException iae) {return ExitCode.INVALID_CONF;}// 1. building the component stack: 构建componnet stackLifecycleComponent server;try {server = buildBookieServer(new BookieConfiguration(conf));} catch (Exception e) {log.error("Failed to build bookie server", e);return ExitCode.SERVER_EXCEPTION;}// 2. start the server， 启动servertry {ComponentStarter.startComponent(server).get();} catch (InterruptedException ie) {Thread.currentThread().interrupt();// the server is interruptedlog.info("Bookie server is interrupted. Exiting ...");} catch (ExecutionException ee) {log.error("Error in bookie shutdown", ee.getCause());return ExitCode.SERVER_EXCEPTION;}return ExitCode.OK;
}

根据注释内容，逻辑是比较清晰的：

解析命令行参数，并将参数保存到ServerConfiguration里；
创建LifecycleComponent，实际类型是一个LifecycleComponentStack，LifecycleComponentStack中会保存多个LifecycleComponent的实例，通过LifecycleComponentStack统一进行启动、停止以及停止的操作，这里保存的每一个LifecycleComponent都是一种服务；
启动所有的服务

下面逐步分析每一步的流程

解析命令行参数

命令行解析的内容主要在parseCommandLine()方法，实际执行解析的过程是在parseArgs()方法：

 private static ServerConfiguration parseArgs(String[] args)throws IllegalArgumentException {try {BasicParser parser = new BasicParser();CommandLine cmdLine = parser.parse(BK_OPTS, args);if (cmdLine.hasOption('h')) {throw new IllegalArgumentException();}ServerConfiguration conf = new ServerConfiguration();if (cmdLine.hasOption('c')) {String confFile = cmdLine.getOptionValue("c");loadConfFile(conf, confFile);}if (cmdLine.hasOption("withAutoRecovery")) {conf.setAutoRecoveryDaemonEnabled(true);}if (cmdLine.hasOption("r")) {conf.setForceReadOnlyBookie(true);}boolean overwriteMetadataServiceUri = false;String sZkLedgersRootPath = "/ledgers";if (cmdLine.hasOption('m')) {sZkLedgersRootPath = cmdLine.getOptionValue('m');log.info("Get cmdline zookeeper ledger path: {}", sZkLedgersRootPath);overwriteMetadataServiceUri = true;}String sZK = conf.getZkServers();if (cmdLine.hasOption('z')) {sZK = cmdLine.getOptionValue('z');log.info("Get cmdline zookeeper instance: {}", sZK);overwriteMetadataServiceUri = true;}// command line arguments overwrite settings in configuration fileif (overwriteMetadataServiceUri) {String metadataServiceUri = "zk://" + sZK + sZkLedgersRootPath;conf.setMetadataServiceUri(metadataServiceUri);log.info("Overwritten service uri to {}", metadataServiceUri);}if (cmdLine.hasOption('p')) {String sPort = cmdLine.getOptionValue('p');log.info("Get cmdline bookie port: {}", sPort);Integer iPort = Integer.parseInt(sPort);conf.setBookiePort(iPort.intValue());}if (cmdLine.hasOption('j')) {String sJournalDir = cmdLine.getOptionValue('j');log.info("Get cmdline journal dir: {}", sJournalDir);conf.setJournalDirName(sJournalDir);}if (cmdLine.hasOption('i')) {String[] sIndexDirs = cmdLine.getOptionValues('i');log.info("Get cmdline index dirs: ");for (String index : sIndexDirs) {log.info("indexDir : {}", index);}conf.setIndexDirName(sIndexDirs);}if (cmdLine.hasOption('l')) {String[] sLedgerDirs = cmdLine.getOptionValues('l');log.info("Get cmdline ledger dirs: ");for (String ledger : sLedgerDirs) {log.info("ledgerdir : {}", ledger);}conf.setLedgerDirNames(sLedgerDirs);}return conf;} catch (ParseException e) {log.error("Error parsing command line arguments : ", e);throw new IllegalArgumentException(e);}}

解析的最终结果是将配置信息保存在ServerConfiguration中，这里的逻辑比较清晰，不再赘述，只需要注意一点，命令行传递的参数优先级高于配置文件配置的参数。

构建bookie所需的服务

这个过程是将bookie 所需的服务统一添加到LifecycleComponentStack中，然后通过LifecycleComponentStack统一启动。

 public static LifecycleComponentStack buildBookieServer(BookieConfiguration conf) throws Exception {LifecycleComponentStack.Builder serverBuilder = LifecycleComponentStack.newBuilder().withName("bookie-server");// 1. build stats provider，指标服务StatsProviderService statsProviderService =new StatsProviderService(conf);StatsLogger rootStatsLogger = statsProviderService.getStatsProvider().getStatsLogger("");serverBuilder.addComponent(statsProviderService);log.info("Load lifecycle component : {}", StatsProviderService.class.getName());// 2. build bookie server，bookie服务BookieService bookieService =new BookieService(conf, rootStatsLogger);serverBuilder.addComponent(bookieService);log.info("Load lifecycle component : {}", BookieService.class.getName());//一致性检查任务if (conf.getServerConf().isLocalScrubEnabled()) {serverBuilder.addComponent(new ScrubberService(rootStatsLogger.scope(ScrubberStats.SCOPE),conf, bookieService.getServer().getBookie().getLedgerStorage()));}// 3. build auto recovery，自动恢复服务if (conf.getServerConf().isAutoRecoveryDaemonEnabled()) {AutoRecoveryService autoRecoveryService =new AutoRecoveryService(conf, rootStatsLogger.scope(REPLICATION_SCOPE));serverBuilder.addComponent(autoRecoveryService);log.info("Load lifecycle component : {}", AutoRecoveryService.class.getName());}// 4. build http service，rest服务if (conf.getServerConf().isHttpServerEnabled()) {BKHttpServiceProvider provider = new BKHttpServiceProvider.Builder().setBookieServer(bookieService.getServer()).setServerConfiguration(conf.getServerConf()).setStatsProvider(statsProviderService.getStatsProvider()).build();HttpService httpService =new HttpService(provider, conf, rootStatsLogger);serverBuilder.addComponent(httpService);log.info("Load lifecycle component : {}", HttpService.class.getName());}// 5. build extra services，其他服务String[] extraComponents = conf.getServerConf().getExtraServerComponents();if (null != extraComponents) {try {List<ServerLifecycleComponent> components = loadServerComponents(extraComponents,conf,rootStatsLogger);for (ServerLifecycleComponent component : components) {serverBuilder.addComponent(component);log.info("Load lifecycle component : {}", component.getClass().getName());}} catch (Exception e) {if (conf.getServerConf().getIgnoreExtraServerComponentsStartupFailures()) {log.info("Failed to load extra components '{}' - {}. Continuing without those components.",StringUtils.join(extraComponents), e.getMessage());} else {throw e;}}}return serverBuilder.build();}

上述代码可以看到：

首先创建 StatsProviderService，这个服务的主要目的是采集指标信息，默认是一个空的实现
创建BookieService 服务，会传入第一步StatsProviderService 的 StatsLogger，然后会初始化一个BookieServer
如果isLocalScrubEnabled为true的话，开启一个本地一致性检查的服务
创建http服务
创建配置中指定的其他的服务

构建状态（指标）服务

// 1. build stats provider
StatsProviderService statsProviderService =new StatsProviderService(conf);
StatsLogger rootStatsLogger = statsProviderService.getStatsProvider().getStatsLogger("");...public StatsProviderService(BookieConfiguration conf) throws Exception {//指定component的名称、配置信息以及StatsLogger，这里 StatsLogger 是 NullStatsLoggersuper(NAME, conf, NullStatsLogger.INSTANCE)S//根据配置配置文件内容初始化 StatsProvider, 默认是 NullStatsProviderClass<? extends StatsProvider> statsProviderClass =conf.getServerConf().getStatsProviderClass();this.statsProvider = ReflectionUtils.newInstance(statsProviderClass);
}

可以看到一个StatsProviderService都有一个名称、StatsLogger以及StatsProvider。下面介绍一下这两个类：

StatsProvider的作用是为不同的scope提供不同的 StatsLogger。

public interface StatsProvider {// 初始化providewrvoid start(Configuration conf);// 停止providervoid stop();// 向writer写入指标信息default void writeAllMetrics(Writer writer) throws IOException {throw new UnsupportedOperationException("writeAllMetrics is not implemented yet");}// 获取指定scope的stats loggerStatsLogger getStatsLogger(String scope);/*** Return the fully qualified stats name comprised of given <tt>statsComponents</tt>.** @param statsComponents stats components to comprise the fully qualified stats name* @return the fully qualified stats name*/default String getStatsName(String...statsComponents) {return StringUtils.join(statsComponents, '/');}
}

StatsLogger的作用是对外提供两个有用的接口，一个是为 OpState提供 OpStatsLogger，一个是为stat提供StatsLogger。

public interface StatsLogger {// 为name指定的OpStat提供loggerOpStatsLogger getOpStatsLogger(String name);// 获取name指定的Counter指标Counter getCounter(String name);// 注册一个Gauge指标<T extends Number> void registerGauge(String name, Gauge<T> gauge);// 卸载一个Gauge指标<T extends Number> void unregisterGauge(String name, Gauge<T> gauge);// 获取name指定的StatsLoggerStatsLogger scope(String name);// 移除scope对应的StatsLoggervoid removeScope(String name, StatsLogger statsLogger);}

构建BookieService

启动BookieService，指定component的名称、配置、以及一个从StatsProvider中获取的StatsLogger。

public BookieService(BookieConfiguration conf,StatsLogger statsLogger)throws Exception {super(NAME, conf, statsLogger);this.server = new BookieServer(conf.getServerConf(), statsLogger);
}

接下来构造一个BookieServer，这是BookKeeper服务的核心。

public BookieServer(ServerConfiguration conf, StatsLogger statsLogger)throws IOException, KeeperException, InterruptedException,BookieException, UnavailableException, CompatibilityException, SecurityException {this.conf = conf;// 验证用户是否在配置中有权限      validateUser(conf);String configAsString;try {configAsString = conf.asJson();LOG.info(configAsString);} catch (ParseJsonException pe) {LOG.error("Got ParseJsonException while converting Config to JSONString", pe);}// 初始化内存分配器ByteBufAllocator allocator = getAllocator(conf);this.statsLogger = statsLogger;// 初始化NettyServer      this.nettyServer = new BookieNettyServer(this.conf, null, allocator);try {// 初始化Bookiethis.bookie = newBookie(conf, allocator);} catch (IOException | KeeperException | InterruptedException | BookieException e) {// interrupted on constructing a bookiethis.nettyServer.shutdown();throw e;}final SecurityHandlerFactory shFactory;shFactory = SecurityProviderFactoryFactory.getSecurityProviderFactory(conf.getTLSProviderFactoryClass());this.requestProcessor = new BookieRequestProcessor(conf, bookie,statsLogger.scope(SERVER_SCOPE), shFactory, bookie.getAllocator());// 为 NettyServer 指定RequestProcessor      this.nettyServer.setRequestProcessor(this.requestProcessor);
}

这里的流程比较长：

校验用户权限
构造内存分配器
构造NettyServer
构造Bookie实例
为NettyServer指定RequestProcessor

这里跳过用户权限校验，重点说明一下后续的部分。

构造内存分配器

private ByteBufAllocator getAllocator(ServerConfiguration conf) {return ByteBufAllocatorBuilder.create()// 分配 buffer 的策略.poolingPolicy(conf.getAllocatorPoolingPolicy())// 分配器的池化并行度.poolingConcurrency(conf.getAllocatorPoolingConcurrency())// oom是的策略，包括两种一种是直接抛出异常，一种退化到heap.outOfMemoryPolicy(conf.getAllocatorOutOfMemoryPolicy())// oom listener，分配失败时，打印提示信息.outOfMemoryListener((ex) -> {try {LOG.error("Unable to allocate memory, exiting bookie", ex);} finally {if (uncaughtExceptionHandler != null) {uncaughtExceptionHandler.uncaughtException(Thread.currentThread(), ex);}}})// Netty 内存泄露的探测策略.leakDetectionPolicy(conf.getAllocatorLeakDetectionPolicy()).build();
}
---
// 定义分配buffer的策略
public enum PoolingPolicy {// 从Heap区分配，不池化。// 这个策略内存占用最小，因为在heap上，JVM GC会自动回收内存，但是可能会影响吞吐UnpooledHeap,// 分配buffer是使用直接内存，并且池化。// 直接内存可以避免GC的消耗，并且可以避免读写channel的内存拷贝。// 池化会增加内存空间的消耗，每个线程为了避免竞争会保存一部分内存作为thread-local内存，分配器汇中存在内存碎片PooledDirect
}
---
// 定义内存泄漏的探测策略
public enum LeakDetectionPolicy {// 不探测，没有开销Disabled,// 使用已分配buffer的%1来跟踪泄露Simple,// 使用已分配buffer的%1来跟踪泄露，并且报告buffer使用未知的堆栈信息Advanced,// 使用已分配buffer的%100来跟踪泄露，并且报告buffer使用未知的堆栈信息, 消耗比较高Paranoid,
}

构造NettyServer

构造NettyServer之前，初始化eventLoopGroup和监听端口，然后就是Netty的标准ServerBootstrap的配置流程。

private void listenOn(InetSocketAddress address, BookieSocketAddress bookieAddress) throws InterruptedException {if (!conf.isDisableServerSocketBind()) {ServerBootstrap bootstrap = new ServerBootstrap();// 设置Accept事件循环和work事件循环中的内存分配器bootstrap.option(ChannelOption.ALLOCATOR, allocator);bootstrap.childOption(ChannelOption.ALLOCATOR, allocator);// 设置Accept事件循环组和work事件循环组bootstrap.group(eventLoopGroup, eventLoopGroup);// 是否立即发送bootstrap.childOption(ChannelOption.TCP_NODELAY, conf.getServerTcpNoDelay());// Timeout to drain the socket on close. 即socket 关闭时可以等待的时间bootstrap.childOption(ChannelOption.SO_LINGER, conf.getServerSockLinger());// 设置自适应的ReceiveBuffer, 真正分配buffer时也是用的上面的allocatorbootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,new AdaptiveRecvByteBufAllocator(conf.getRecvByteBufAllocatorSizeMin(),conf.getRecvByteBufAllocatorSizeInitial(), conf.getRecvByteBufAllocatorSizeMax()));// 高低水位，用于限制客户端限流bootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(conf.getServerWriteBufferLowWaterMark(), conf.getServerWriteBufferHighWaterMark()));if (eventLoopGroup instanceof EpollEventLoopGroup) {bootstrap.channel(EpollServerSocketChannel.class);} else {bootstrap.channel(NioServerSocketChannel.class);}// 添加channel 添加 hanlder pipelinebootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {synchronized (suspensionLock) {while (suspended) {suspensionLock.wait();}}BookieSideConnectionPeerContextHandler contextHandler =new BookieSideConnectionPeerContextHandler();ChannelPipeline pipeline = ch.pipeline();// For ByteBufList, skip the usual LengthFieldPrepender and have the encoder itself to add it// Outboundpipeline.addLast("bytebufList", ByteBufList.ENCODER_WITH_SIZE);// Inboundpipeline.addLast("lengthbaseddecoder", new LengthFieldBasedFrameDecoder(maxFrameSize, 0, 4, 0, 4));// Outboundpipeline.addLast("lengthprepender", new LengthFieldPrepender(4));// Inboundpipeline.addLast("bookieProtoDecoder", new BookieProtoEncoding.RequestDecoder(registry));// Outboundpipeline.addLast("bookieProtoEncoder", new BookieProtoEncoding.ResponseEncoder(registry));// Inboundpipeline.addLast("bookieAuthHandler", new AuthHandler.ServerSideHandler(contextHandler.getConnectionPeer(), authProviderFactory));// 正常情况下，channel初始化时，isRunning会被设置成true，并且requestProcessor// 会被设置为BookieRequestProcessorChannelInboundHandler requestHandler = isRunning.get()? new BookieRequestHandler(conf, requestProcessor, allChannels): new RejectRequestHandler();// Inboundpipeline.addLast("bookieRequestHandler", requestHandler);// Inboundpipeline.addLast("contextHandler", contextHandler);}});// Bind and start to accept incoming connectionsChannel listen = bootstrap.bind(address.getAddress(), address.getPort()).sync().channel();if (listen.localAddress() instanceof InetSocketAddress) {if (conf.getBookiePort() == 0) {conf.setBookiePort(((InetSocketAddress) listen.localAddress()).getPort());}}}...}

主要配置是在ChannelInitializer中设置Handler。

构建Bookie

bookie是BK的核心，构建Bookie的过程如下：

protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator)throws IOException, KeeperException, InterruptedException, BookieException {return conf.isForceReadOnlyBookie()? new ReadOnlyBookie(conf, statsLogger.scope(BOOKIE_SCOPE), allocator): new Bookie(conf, statsLogger.scope(BOOKIE_SCOPE), allocator);
}

根据配置指定是ReadOnlyBookie还是普通的bookie，以普通bookie为例：

public Bookie(ServerConfiguration conf, StatsLogger statsLogger, ByteBufAllocator allocator)throws IOException, InterruptedException, BookieException {super("Bookie-" + conf.getBookiePort());this.statsLogger = statsLogger;this.conf = conf;// 从配置文件中获取journal 目录 list，然后在在每个目录下创建一个current目录this.journalDirectories = Lists.newArrayList();for (File journalDirectory : conf.getJournalDirs()) {this.journalDirectories.add(getCurrentDirectory(journalDirectory));}// 初始化DiskChecker，有两个参数 diskUsageThreshold 和 diskUsageWarnThreshold// diskUsageThreshold表示磁盘的最大使用率，默认是0.95，目录列表中的所有目录都超过限制之后// 如果bookie配置可以以readonly模式运行，就会转化为readonly状态，否则会停止；// diskUsageWarnThreshold 表示磁盘使用的告警阈值，默认是0.90，超过这个值会抛出// DiskWarnThresholdException，并且会触发gc，当使用率低于这个值时，目录重新变为开写状态DiskChecker diskChecker = createDiskChecker(conf);// 为ledger和index创建LedgerDirsManager，用来管理ledger和index的目录列表this.ledgerDirsManager = createLedgerDirsManager(conf, diskChecker, statsLogger.scope(LD_LEDGER_SCOPE));this.indexDirsManager = createIndexDirsManager(conf, diskChecker, statsLogger.scope(LD_INDEX_SCOPE), this.ledgerDirsManager);this.allocator = allocator;// 初始化zk 客户端this.metadataDriver = instantiateMetadataDriver(conf);checkEnvironment(this.metadataDriver);try {if (this.metadataDriver != null) {// 初始化ledgerManagerFactory，用于生成ledgerManagerledgerManagerFactory = metadataDriver.getLedgerManagerFactory();LOG.info("instantiate ledger manager {}", ledgerManagerFactory.getClass().getName());// ledgerManager负责和zk等元数据存储交互，用来管理ledger的元数据信息ledgerManager = ledgerManagerFactory.newLedgerManager();} else {ledgerManagerFactory = null;ledgerManager = null;}} catch (MetadataException e) {throw new MetadataStoreException("Failed to initialize ledger manager", e);}// 初始化状态管理器stateManager = initializeStateManager();// register shutdown handler using trigger modestateManager.setShutdownHandler(exitCode -> triggerBookieShutdown(exitCode));// LedgerDirsMonitor, 监控所有配置的目录，如果发现磁盘错误或者所有的leger 目录都满，就抛出异常，// bookie启动失败List<LedgerDirsManager> dirsManagers = new ArrayList<>();dirsManagers.add(ledgerDirsManager);if (indexDirsManager != ledgerDirsManager) {dirsManagers.add(indexDirsManager);}this.dirsMonitor = new LedgerDirsMonitor(conf, diskChecker, dirsManagers);try {this.dirsMonitor.init();} catch (NoWritableLedgerDirException nle) {// start in read-only mode if no writable dirs and read-only allowedif (!conf.isReadOnlyModeEnabled()) {throw nle;} else {this.stateManager.transitionToReadOnlyMode();}}// instantiate the journals， 初始化journaljournals = Lists.newArrayList();for (int i = 0; i < journalDirectories.size(); i++) {journals.add(new Journal(i, journalDirectories.get(i),conf, ledgerDirsManager, statsLogger.scope(JOURNAL_SCOPE), allocator));}this.entryLogPerLedgerEnabled = conf.isEntryLogPerLedgerEnabled();CheckpointSource checkpointSource = new CheckpointSourceList(journals);// 初始化ledgerStore，默认是一个 SortedLedgerStorageledgerStorage = buildLedgerStorage(conf);boolean isDbLedgerStorage = ledgerStorage instanceof DbLedgerStorage;/** with this change https://github.com/apache/bookkeeper/pull/677,* LedgerStorage drives the checkpoint logic.** <p>There are two exceptions:** 1) with multiple entry logs, checkpoint logic based on a entry log is*    not possible, hence it needs to be timebased recurring thing and*    it is driven by SyncThread. SyncThread.start does that and it is*    started in Bookie.start method.** 2) DbLedgerStorage*/// 一般都是由 LedgerStorage来驱动checkpoint 逻辑，但是有两个例外：// 1. 有多个entry logs，cp逻辑不能依赖于一个entry log，应该是一个基于时间的循环，有SyncThread驱动// 2. DbLegerStorageif (entryLogPerLedgerEnabled || isDbLedgerStorage) {syncThread = new SyncThread(conf, getLedgerDirsListener(), ledgerStorage, checkpointSource) {@Overridepublic void startCheckpoint(Checkpoint checkpoint) {/** in the case of entryLogPerLedgerEnabled, LedgerStorage* dont drive checkpoint logic, but instead it is done* periodically by SyncThread. So startCheckpoint which* will be called by LedgerStorage will be no-op.*/}@Overridepublic void start() {executor.scheduleAtFixedRate(() -> {doCheckpoint(checkpointSource.newCheckpoint());}, conf.getFlushInterval(), conf.getFlushInterval(), TimeUnit.MILLISECONDS);}};} else {syncThread = new SyncThread(conf, getLedgerDirsListener(), ledgerStorage, checkpointSource);}// 初始化LedgerStorageledgerStorage.initialize(conf,ledgerManager,ledgerDirsManager,indexDirsManager,stateManager,checkpointSource,syncThread,statsLogger,allocator);// HandleFactoryImpl 用来获取 handle，这里的 hanlde 是 LedgerDescriptor，是 ledger 的实现// 主要负责向 ledger addEntry 和或者从ledger readeEntryhandles = new HandleFactoryImpl(ledgerStorage);// Expose Statsthis.bookieStats = new BookieStats(statsLogger);
}

这里的逻辑比较多，主要包括内容是：

磁盘管理：为journal、ledger和index目录创建 LedgerDirsMonitor，用来监控目录的使用情况
元数据管理：创建ZK 客户端，用于ledger 元数据的管理
数据存储管理：初始化ledgerStore，用于将ledger数据持久化到磁盘
SyncThread：初始化SyncTread，用于数据刷盘

BookKeeper源码解析之Bookie启动流程（一）相关推荐

kotlin coroutine源码解析之Job启动流程
目录 Job启动流程 launch流程分析父子Job关联分析结论 Job启动流程 job启动流程,我们先从一段最简单使用协程的代码开始,进行代码跟跟踪,顺便引出几个关键的概念,在后面章节里面去单独 ...
基于8.0源码解析：startService 启动过程
基于8.0源码解析:startService 启动过程首先看一张startService的图,心里有个大概的预估,跟Activity启动流程比,Service的启动稍微简单点,并且我把Service ...
12.源码阅读（app启动流程-android api 26）
activity的启动流程之前已经通过源码了解了,那么app的启动流程是怎样的,从我们按下app的图标,到应用启动起来显示出画面,中间都经历了什么? 安卓是基于java的,所以和java有一定的相似性 ...
详细讲解go web框架之gin框架源码解析记录及思路流程和理解
开篇首先gin 框架是在官方提供的net/http标准包进行的相应封装. 那么要想理解gin框架, 就要先懂一些 net/http标准包的相关知识. 可以参考中文的文档: https://st ...
rocketmq源码解析之name启动（一）
2019独角兽企业重金招聘Python工程师标准>>> 说在前面主要解析namrsrv启动部分,namesrv配置加载.netty server创建.注册出处理器. 正文源码解析 ...
源码解析-偏向锁撤销流程解读
一.单个偏向锁的撤销源码链接:http://hg.openjdk.java.net/jdk8u/jdk8u/hotspot/file/9ce27f0a4683/src/share/vm/runtim ...
6、RocketMQ 源码解析之 Broker 启动(上)
上面一篇我们介绍了 RocketMQ 的元数据管理,它是通过自定义一个 KV 服务器.并且其它服务在 NameServer 注册服务信息的时候都是全量注册.如果 RocketMQ 的拓扑图当中有多台 ...
Glide 4.9源码解析-图片加载流程
本文Glide源码基于4.9,版本下载地址如下:Glide 4.9 前言由于Glide源码真的很复杂,因此本文只分析和贴出与图片加载流程相关的功能以及代码.另外本文Glide源码基于4.9,与3.x ...
源码分析Dubbo服务提供者启动流程-下篇
本文继续上文Dubbo服务提供者启动流程,在上篇文章中详细梳理了从dubbo spring文件开始,Dubbo是如何加载配置文件,服务提供者dubbo:service标签服务暴露全流程,本节重点关注R ...

BookKeeper源码解析之Bookie启动流程（一）