apache bookkeeper bookie 启动流程源码分析

java

Apache BookKeeper 是一个可以方便扩展，高可用，低延迟的存储系统。BookKeeper 专门为 append-only 的工作模式提供了优化，在以下的应用场景中非常适用：

  • WAL (Write-Ahead-Logging)， 例如 HDFS 的 NameNode

  • 消息存储系统，例如 Apache Pulsar

  • Offset/Cursor 存储系统，例如在 Apache Pulsar 中用来存储消息消费位置

  • Object/Blob Store 对象存储系统，例如存储状态机的 snapshots

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
public static void main(String[] args) {
        int retCode = doMain(args);
        Runtime.getRuntime().exit(retCode);
    }

    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:
        LifecycleComponent 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
        try {
            ComponentStarter.startComponent(server).get();
        } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();
            // the server is interrupted
            log.info("Bookie server is interrupted. Exiting ...");
        } catch (ExecutionException ee) {
            log.error("Error in bookie shutdown", ee.getCause());
            return ExitCode.SERVER_EXCEPTION;
        }
        return ExitCode.OK;
    }

注释很清晰，可以看到启动bookie服务主要做这三步：

  • 解析system property
  • 构建bookie服务所需的组件
  • 启动各个组件

解析system property

1
2
BasicParser parser = new BasicParser();
CommandLine cmdLine = parser.parse(BK_OPTS, args);

system property是java应用程序自身指定的变量，通常我们可以在启动应用的时候指定的，格式是:-DsystemPropertyKey=systemPropertyValue(楼主在本地启动bookie服务在idea设置的Program rguments：–conf /Volumes/longmao/bookkeeper-confg/b1.conf)，解析system property使用了apache开源工具commons-cli(自己写应用或框架可以借鉴下其写法，用来加载应用自定义的配置)。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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");
    }

含义：

  • -c/–conf:使用的配置文件
  • -r/–readOnly:是否只读

构建bookie服务所需的组件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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
        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
        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();
    }

利用第一步解析配置生成的BookieConfiguration对象构造bookie服务依赖的组件，bookie启动过程中需要启动的服务组件：

  • StatsProviderService 指标服务
  • BookieService bookie服务
  • AutoRecoveryService 自动恢复服务
  • HttpService http rest服务
  • 其它服务

利用了LifecycleComponentStack这各类保存了需要启动的组件并规定了组件生命周期内执行的方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Override
    public void start() {
        components.forEach(component -> component.start());
    }

    @Override
    public void stop() {
        components.reverse().forEach(component -> component.stop());
    }

    @Override
    public void close() {
        components.reverse().forEach(component -> component.close());
    }

有没有觉得很熟悉，tomcat源码里也有类似的设计，tomcat中的接口：org.apache.catalina.Lifecycle 定义了容器生命周期、容器状态转换及容器状态迁移事件的监听器和移除等主要接口，tomcat里的组件StandardHost等实现了这个接口，维护自身生命周期运转过程中要执行的逻辑。只能说优秀的源码套路都是差不多^_^

启动服务

遍历LifecycleComponentStack中的
ImmutableList components对象，执行各个服务组件的start方法