JStorm 源码解析：nimbus 的启动和运行机制

private void launchServer(final Map conf, INimbus inimbus) {
    LOG.info("Begin to start nimbus with conf " + conf);
    try {
        // 1. 验证当前为分布式运行模式，不允许以本地模式运行
        StormConfig.validate_distributed_mode(conf);

        // 2. 创建当前 JVM 进程对应的目录：${storm.local.dir}/nimbus/pids/${pid}，如果存在历史运行记录，则会进行清除
        this.createPid(conf);

        // 3. 注册 shutdown hook 方法，用于在 JVM 进程终止时执行清理逻辑
        this.initShutdownHook();

        // 4. 模板方法
        inimbus.prepare(conf, StormConfig.masterInimbus(conf));

        // 5. 基于 conf 创建 NimbusData 对象
        data = this.createNimbusData(conf, inimbus);

        // 6. 注册一个 follower 线程
        this.initFollowerThread(conf);

        // 7. 创建并启动一个后端 HTTP 服务（默认端口为 7621，主要用于查看和下载 nimbus 的日志数据）
        int port = ConfigExtension.getNimbusDeamonHttpserverPort(conf);
        hs = new Httpserver(port, conf);
        hs.start();

        // 8. 如果集群运行在 YARN 上，则初始化容器心跳线程
        this.initContainerHBThread(conf);

        // 9. 创建 ServiceHandler（实现了 Nimbus.Iface），并启动 Thrift 服务，用于处理 Nimbus 请求
        serviceHandler = new ServiceHandler(data);
        this.initThrift(conf);
    } catch (Throwable e) {
        if (e instanceof OutOfMemoryError) {
            LOG.error("Halting due to out of memory error...");
        }
        LOG.error("Fail to run nimbus ", e);
    } finally {
        this.cleanup();
    }
    LOG.info("Quit nimbus");
}

public void createFileHandler() {
    // 注册一个 callback 方法，基于回调的方式关闭管道或输入流
    ExpiredCallback<Object, Object> expiredCallback = new ExpiredCallback<Object, Object>() {
        @Override
        public void expire(Object key, Object val) {
            try {
                LOG.info("Close file " + String.valueOf(key));
                if (val != null) {
                    if (val instanceof Channel) {
                        Channel channel = (Channel) val;
                        channel.close();
                    } else if (val instanceof BufferFileInputStream) {
                        BufferFileInputStream is = (BufferFileInputStream) val;
                        is.close();
                    }
                }
            } catch (IOException e) {
                LOG.error(e.getMessage(), e);
            }

        }
    };

    /*
     * 获取文件上传和下载的超时时间，默认为 30 秒
     *
     * During upload/download with the master,
     * how long an upload or download connection is idle before nimbus considers it dead and drops the connection.
     */
    int file_copy_expiration_secs = JStormUtils.parseInt(conf.get(Config.NIMBUS_FILE_COPY_EXPIRATION_SECS), 30);

    /*
     * TimeCacheMap 在实例化时会启动一个守护线程，
     * 并依据超时时间循环从 buckets 中去除对象，并应用执行 callback 的 expire 方法
     * 这里的 expire 逻辑是执行关闭管道或输入流
     */
    uploaders = new TimeCacheMap<>(file_copy_expiration_secs, expiredCallback);
    downloaders = new TimeCacheMap<>(file_copy_expiration_secs, expiredCallback);
}

public TimeCacheMap(int expirationSecs, int numBuckets, ExpiredCallback<K, V> callback) {
    if (numBuckets < 2) {
        throw new IllegalArgumentException("numBuckets must be >= 2");
    }
    buckets = new LinkedList<>();
    for (int i = 0; i < numBuckets; i++) {
        buckets.add(new HashMap<K, V>());
    }

    // 注册回调策略
    this.callback = callback;
    final long expirationMillis = expirationSecs * 1000L;
    final long sleepTime = expirationMillis / (numBuckets - 1);

    /*
     * cleaner 线程会一直循环的执行，
     * 间隔指定时间从缓冲区尾部获取对象，并为该对象应用 callback 的 expire 方法
     */
    this.cleaner = new Thread(new Runnable() {

        @Override
        public void run() {
            while (!AsyncLoopRunnable.getShutdown().get()) {
                Map<K, V> dead;
                JStormUtils.sleepMs(sleepTime);
                synchronized (lock) {
                    // 从缓冲区队尾获取对象
                    dead = buckets.removeLast();
                    // 添加一个空的 map 到缓冲区，从而保证线程的正常运行
                    buckets.addFirst(new HashMap<K, V>());
                }
                if (TimeCacheMap.this.callback != null) {
                    for (Entry<K, V> entry : dead.entrySet()) {
                        TimeCacheMap.this.callback.expire(entry.getKey(), entry.getValue());
                    }
                }
            }
        }
    });
    cleaner.setDaemon(true);
    cleaner.start();
}

- supervisors
- topology
- assignments
- assignments_bak
- tasks
- taskbeats
- taskerrors
- metrics
- blobstore
- gray_upgrade

+ ${zk_root_dir}
| ---- + topology: 记录集群中所有正在运行的 topology 数据
| ---- | ---- + ${topology_id}: 指定 topology 的相关信息（名称、开始运行时间、运行状态等）

| ---- + supervisors: 记录集群中所有 supervisor 节点的心跳信息
| ---- | ---- + ${supervivor_id}: 指定 supervisor 的心跳信息（心跳时间、主机名称、所有 worker 的端口号、运行时间等）

| ---- + assignments: 记录提交给集群的 topology 任务分配信息
| ---- | ---- + ${topology_id}: 指定 topology 的任务分配信息（对应 nimbus 上的代码目录、所有 task 的启动时间、每个 task 与节点和端口的映射关系等）

| ---- + assignments_bak: 记录提交给集群的 topology 任务分配信息的备份

| ---- + tasks: 记录集群中所有 topology 的 task 信息
| ---- | ---- + ${topology_id}: 指定 topology 的所有 task 信息
| ---- | ---- | ---- + ${task_id}: 指定 task 所属的组件 ID 和类型（spout/bolt）

| ---- + taskbeats: 记录集群中所有 task 的心跳信息
| ---- | ---- + ${topology_id}: 记录指定 topology 下所有 task 的心跳信息、topologyId，以及 topologyMasterId 等
| ---- | ---- | ---- + ${task_id}: 指定 task 的心跳信息（最近一次心跳时间、运行时长、统计信息等）

| ---- + taskerrors: 记录集群中所有 topology 的 task 运行错误信息
| ---- | ---- + ${topology_id}: 指定 topology 下所有 task 的运行错误信息
| ---- | ---- | ---- + ${task_id}: 指定 task 的运行错误信息

| ---- + metrics: 记录集群中所有 topology 的 metricsId

| ---- + blobstore: 记录集群对应的 blobstore 信息，用于协调数据一致性

| ---- + gray_upgrade: 记录灰度发布中的 topologyId

public NimbusCache(Map conf, StormClusterState zkCluster) {
    super();

    // 获取本地缓存的具体实现类
    String dbCacheClass = this.getNimbusCacheClass(conf);
    LOG.info("NimbusCache db cache will use {}", dbCacheClass);

    try {
        dbCache = (JStormCache) Utils.newInstance(dbCacheClass);

        String dbDir = StormConfig.masterDbDir(conf);
        // 设置本地缓存数据存放目录
        conf.put(RocksDBCache.ROCKSDB_ROOT_DIR, dbDir); // ${storm.local.dir}/nimbus/rocksdb
        // 是否在 nimbus 启动时清空数据，默认为 true
        conf.put(RocksDBCache.ROCKSDB_RESET, ConfigExtension.getNimbusCacheReset(conf));
        dbCache.init(conf);
        if (dbCache instanceof TimeoutMemCache) {
            memCache = dbCache;
        } else {
            memCache = new TimeoutMemCache();
            memCache.init(conf);
        }
    }
    // 省略 catch 代码块
    this.zkCluster = zkCluster;
}

private void initFollowerThread(Map conf) {
    // 如果当前 nimbus 成为 leader，则会触发此回调执行初始化操作
    Callback leaderCallback = new Callback() {
        @Override
        public <T> Object execute(T... args) {
            try {
                init(data.getConf());
            } catch (Exception e) {
                LOG.error("Nimbus init error after becoming a leader", e);
                JStormUtils.halt_process(0, "Failed to init nimbus");
            }
            return null;
        }
    };
    // 创建并启动 follower 线程
    follower = new FollowerRunnable(data, 5000, leaderCallback);
    Thread thread = new Thread(follower);
    thread.setDaemon(true);
    thread.start();
    LOG.info("Successfully init Follower thread");
}

private void tryToBeLeader(final Map conf) throws Exception {
    // 依据候选 nimbus 从节点的优先级来决定当前 nimbus 从节点是否有资格尝试成为 leader
    boolean allowed = this.check_nimbus_priority();
    if (allowed) {
        // 回调策略再次尝试
        RunnableCallback masterCallback = new RunnableCallback() {
            @Override
            public void run() {
                try {
                    tryToBeLeader(conf);
                } catch (Exception e) {
                    LOG.error("tryToBeLeader error", e);
                    JStormUtils.halt_process(30, "Cant't be master" + e.getMessage());
                }
            }
        };
        // 尝试成为 leader 节点
        LOG.info("This nimbus can be leader");
        data.getStormClusterState().try_to_be_leader(Cluster.MASTER_SUBTREE, hostPort, masterCallback);
    } else {
        LOG.info("This nimbus can't be leader");
    }
}

+ ${nimbus_local_dir}
| ---- + nimbus
| ---- | ---- + inbox: 存放客户端上传的 jar 包
| ---- | ---- | ---- + stormjar-{uuid}.jar: 对应一个具体的 jar 包
| ---- | ---- + stormdist
| ---- | ---- | ---- + ${topology_id}
| ---- | ---- | ---- | ---- + stormjar.jar: 包含当前拓扑所有代码的 jar 包（从 inbox 那复制过来的）
| ---- | ---- | ---- | ---- + stormcode.ser: 当前拓扑对象的序列化文件
| ---- | ---- | ---- | ---- + stormconf.ser: 当前拓扑的配置信息文件

private void initThrift(Map conf) throws TTransportException {
    // 获取 thrift 端口，默认为 8627
    Integer thrift_port = JStormUtils.parseInt(conf.get(Config.NIMBUS_THRIFT_PORT)); // ${nimbus.thrift.port}
    TNonblockingServerSocket socket = new TNonblockingServerSocket(thrift_port);

    // ${nimbus.thrift.max_buffer_size}
    Integer maxReadBufSize = JStormUtils.parseInt(conf.get(Config.NIMBUS_THRIFT_MAX_BUFFER_SIZE));
    // 设置服务运行参数
    THsHaServer.Args args = new THsHaServer.Args(socket);
    args.workerThreads(ServiceHandler.THREAD_NUM); // 64
    args.protocolFactory(new TBinaryProtocol.Factory(false, true, maxReadBufSize, -1));
    args.processor(new Nimbus.Processor<Iface>(serviceHandler));
    args.maxReadBufferBytes = maxReadBufSize;

    thriftServer = new THsHaServer(args);

    LOG.info("Successfully started nimbus: started Thrift server...");
    thriftServer.serve();
}