Doris FE启动流程源码详细解析

一、简介

Apache Doris是一个现代化的MPP分析型数据库产品。仅需亚秒级响应时间即可获得查询结果，有效地支持实时数据分析。Apache Doris的分布式架构非常简洁，易于运维，并且可以支持10PB以上的超大数据集。

Apache Doris可以满足多种数据分析需求，例如固定历史报表，实时数据分析，交互式数据分析和探索式数据分析等。令您的数据分析工作更加简单高效！

二、名词解释

FE：Frontend，即 Doris 的前端节点。主要负责接收和返回客户端请求、元数据以及集群管理、查询计划生成等工作。
BE：Backend，即 Doris 的后端节点。主要负责数据存储与管理、查询计划执行等工作。
bdbje：Oracle Berkeley DB Java Edition (opens new window)。在 Doris 中，我们使用 bdbje 完成元数据操作日志的持久化、FE 高可用等功能。

三、流程图

在这里插入图片描述

四、源码分析

下载Doris源码详细步骤：https://doris.apache.org/zh-CN/developer-guide/fe-idea-dev.html#_1-%E7%8E%AF%E5%A2%83%E5%87%86%E5%A4%87

Doris FE启动步骤

我们先看看 FE启动类代码：

if (Strings.isNullOrEmpty(dorisHomeDir)) {
        System.err.println("env DORIS_HOME is not set.");
        return;
    }

    if (Strings.isNullOrEmpty(pidDir)) {
        System.err.println("env PID_DIR is not set.");
        return;
    }

    CommandLineOptions cmdLineOpts = parseArgs(args);

    try {
        // 创建 pid 文件
        if (!createAndLockPidFile(pidDir + "/fe.pid")) {
            throw new IOException("pid file is already locked.");
        }

        // 初始化 config文件
        Config config = new Config();
        config.init(dorisHomeDir + "/conf/fe.conf");
        // Must init custom config after init config, separately.
        // Because the path of custom config file is defined in fe.conf
        config.initCustom(Config.custom_config_dir + "/fe_custom.conf");

        LdapConfig ldapConfig = new LdapConfig();
        if (new File(dorisHomeDir + "/conf/ldap.conf").exists()) {
            ldapConfig.init(dorisHomeDir + "/conf/ldap.conf");
        }

        // check it after Config is initialized, otherwise the config 'check_java_version' won't work.
        if (!JdkUtils.checkJavaVersion()) {
            throw new IllegalArgumentException("Java version doesn't match");
        }

        Log4jConfig.initLogging(dorisHomeDir + "/conf/");

        // set dns cache ttl
        java.security.Security.setProperty("networkaddress.cache.ttl", "60");

        // check command line options
        checkCommandLineOptions(cmdLineOpts);

        LOG.info("Palo FE starting...");

        //FE Address 初始化
        FrontendOptions.init();

        // 检查端口是否正常
        checkAllPorts();

        if (Config.enable_bdbje_debug_mode) {
            // Start in BDB Debug mode
            BDBDebugger.get().startDebugMode(dorisHomeDir);
            return;
        }

        // 初始化 Catelog 并且等待加载完成
        Catalog.getCurrentCatalog().initialize(args);
        Catalog.getCurrentCatalog().waitForReady();


        // 第一步 启动 HttpServer 类
        // 第二步 启动 FeServer 类
        // 第三步 启动 QeService
        QeService qeService = new QeService(Config.query_port, Config.mysql_service_nio_enabled, ExecuteEnv.getInstance().getScheduler());
        FeServer feServer = new FeServer(Config.rpc_port);

        feServer.start();

        HttpServer httpServer = new HttpServer();
        httpServer.setPort(Config.http_port);
        httpServer.setMaxHttpPostSize(Config.jetty_server_max_http_post_size);
        httpServer.setAcceptors(Config.jetty_server_acceptors);
        httpServer.setSelectors(Config.jetty_server_selectors);
        httpServer.setWorkers(Config.jetty_server_workers);
        httpServer.setMaxThreads(Config.jetty_threadPool_maxThreads);
        httpServer.setMaxThreads(Config.jetty_threadPool_minThreads);
        httpServer.start();
        
        qeService.start();

        ThreadPoolManager.registerAllThreadPoolMetric();

        while (true) {
            Thread.sleep(2000);
        }
    } catch (Throwable e) {
        e.printStackTrace();
    }

通过上面代码，我们可以清楚了解FE启动时主要执行以下过程：

初始化 Catalog ，并且等待Catelog加载完成
创建 QeServer ，负责与mysql client 通信
创建 FeServer ，由Thrift Server组成，负责 FE 和 BE 通信
创建 HttpServer ，负责提供Rest API以及Doris FE前端页面接口

CataLog 源码解析

CataLog 主要职责是维护FE 元数据，接下来我们看看FE启动时，CataLog初始化时，做什么处理：

// 获取本地节点和helper节点信息
     getSelfHostPort();
     getHelperNodes(args);

     // 检查meta文件目录是否创建
     File meta = new File(metaDir);
     if (!meta.exists()) {
         LOG.warn("Doris' meta dir {} does not exist. You need to create it before starting FE", meta.getAbsolutePath());
         throw new Exception(meta.getAbsolutePath() + " does not exist, will exit");
     }

     //检查 BDB和Image目录是否创建
     if (Config.edit_log_type.equalsIgnoreCase("bdb")) {
         File bdbDir = new File(this.bdbDir);
         if (!bdbDir.exists()) {
             bdbDir.mkdirs();
         }

         File imageDir = new File(this.imageDir);
         if (!imageDir.exists()) {
             imageDir.mkdirs();
         }
     } else {
         throw new Exception("Invalid edit log type: " + Config.edit_log_type);
     }

     // 初始化插件管理
     pluginMgr.init();
     auditEventProcessor.start();

     // 2.获取集群ID和角色(Observer or Follower)
     getClusterIdAndRole();

     // 3. 首次加载image文件和回放Elog日志
     this.editLog = new EditLog(nodeName);
     loadImage(this.imageDir); // 加载image文件
     editLog.open(); // 夹杂bdb环境配置
     this.globalTransactionMgr.setEditLog(editLog);
     this.idGenerator.setEditLog(editLog);

     // 4. 创建加载和导出作业标签清理Daemon线程
     createLabelCleaner();

     // 5. 创建事务清理Daemon线程
     createTxnCleaner();

     // 6. 开始监听线程状态（MASTER/FOLLOWER/OBSERVER状态转换，以及leader选举工作和元数据同步工作）
     createStateListener();
     listener.start();

通过上面源码，我们可以发现，CateLog初始化时，执行以下操作：

首先对Image镜像文件读取数据，和对Elog进行回放操作。
创建加载和导出作业标签清理Daemon线程
创建事务清理Daemon线程
开始监听线程状态（MASTER/FOLLOWER/OBSERVER状态转换，以及leader选举工作和元数据同步工作）（后续更新一篇文章，专门说元数据同步和Leader选举流程源码解析）

QeServer 源码解析

QeServer职责是与Mysql Client进行通讯，支持Socket和Nio连接，具体源码：
```
try {
         HelpModule.getInstance().setUpModule();
     } catch (Exception e) {
         LOG.error("Help module failed, because:", e);
     }
     this.port = port;
     if (nioEnabled) {
         mysqlServer = new NMysqlServer(port, scheduler);
     } else {
         mysqlServer = new MysqlServer(port, scheduler);
     }
```
当nioEnabled（可配置）为true时，使用Nio进行通讯，采用这种方式通信的好处是：
- 同步非阻塞IO
- IO是面向流的，NIO是面向缓冲区的
- NIO引入了选择器的概念，选择器用于监听多个通道的事件
FeServer 源码解析

FeServer职责是负责FE和BE之间通信。

try {
            switch (type) {
                case SIMPLE:
                    createSimpleServer();
                    break;
                case THREADED_SELECTOR:
                    createThreadedServer();
                    break;
                default:
                    createThreadPoolServer();
            }
        } catch (TTransportException ex) {
            LOG.warn("create thrift server failed.", ex);
            throw new IOException("create thrift server failed.", ex);
        }

        ThriftServerEventProcessor eventProcessor = new ThriftServerEventProcessor(this);
        server.setServerEventHandler(eventProcessor);

        serverThread = new Thread(new Runnable() {
            @Override
            public void run() {
                server.serve();
            }
        });
        serverThread.setDaemon(true);
        serverThread.start();

FE的Thrift使用的服务模型分为三种：

SIMPLE：一般不适用于生产环境，仅限于测试使用。
THREADED_SELECTOR：非阻塞式I/O模型，即主从 Reactor 模型，该模型能及时响应大量的并发连接请求，在多数场景下有较好的表现。
THREAD_POOL：阻塞式I/O模型，使用线程池处理用户连接，并发连接数受限于线程池的数量，如果能提前预估并发请求的数量，并且能容忍足够多的线程资源开销，该模型会有较好的性能表现，默认使用该服务模型

HttpServer 源码解析
HttpServer职责主要是为Rest API和doris Web页面提供接口服务，源码如下：

Map<String, Object> properties = new HashMap<>();
    properties.put("server.port", port);
    properties.put("server.servlet.context-path", "/");
    properties.put("spring.resources.static-locations", "classpath:/static");
    properties.put("spring.http.encoding.charset", "UTF-8");
    properties.put("spring.http.encoding.enabled", true);
    properties.put("spring.http.encoding.force", true);
    //enable jetty config
    properties.put("server.jetty.acceptors", this.acceptors);
    properties.put("server.jetty.max-http-post-size", this.maxHttpPostSize);
    properties.put("server.jetty.selectors", this.selectors);
    //Worker thread pool is not set by default, set according to your needs
    if(this.workers > 0) {
        properties.put("server.jetty.workers", this.workers);
    }
    // This is to disable the spring-boot-devtools restart feature.
    // To avoid some unexpected behavior.
    System.setProperty("spring.devtools.restart.enabled", "false");
    // Value of `DORIS_HOME_DIR` is null in unit test.
    if (PaloFe.DORIS_HOME_DIR != null) {
        System.setProperty("spring.http.multipart.location", PaloFe.DORIS_HOME_DIR);
    }
    System.setProperty("spring.banner.image.location", "doris-logo.png");
    if (FeConstants.runningUnitTest) {
        // this is currently only used for unit test
        properties.put("logging.config", getClass().getClassLoader().getResource("log4j2.xml").getPath());
    } else {
        properties.put("logging.config", Config.custom_config_dir + "/" + SpringLog4j2Config.SPRING_LOG_XML_FILE);
    }
    new SpringApplicationBuilder()
            .sources(HttpServer.class)
            .properties(properties)
            .run(new String[]{});

HttpServer继承了SpringBootServletInitializer，同时使用了SpringApplicationBuilder类，那么我们就可以很清楚知道，使用Springboot框架提供Rest Api服务。

五、问题思考

CataLog 如何对Elog进行回放？
ELog 日志如何实现数据同步？
BDB 如何存储元数据的？
说说Doris FE Leader选举流程
Doris FE Leader节点如何判断non-Leader节点是否在线？leader和non-leader长时间失联会导致non-leader提供过期元数据，此问题如何解决？