const worker = childProcess.spawn(cli, args);
// 监听控制台日志输出
worker.stdout.on("data", (message) => {
  // 构造消息体
  const logLine: ILogLine = {
    date: new Date().getTime(),
    level: LogLevel.INFO,
    message: message.toString(),
  };
  /** 收集10条后上报日志 */
});

// 监听控制台错误日志输出
worker.stderr.on("data", (message) => {
  // 构造消息体
  const logLine: ILogLine = {
    date: new Date().getTime(),
    level: LogLevel.ERROR,
    message: message.toString(),
  };
  /** 上报日志 */
});

// 构建结束
worker.on("exit", () => {
  /** 收集所有日志 */
});

• 子线程 worker 中会有3 个监听器，由于分布在不同的监听器，如果想收集所有日志的话，需要自己构建一个队列

• 为了减少请求次数，日志会在收集指定条数后才会上报一次，但是由于是在不同的监听器收集，所以所有的单条日志都需要先 push 到队列中

• 队列在满10条之后会取出进行日志的批量上报

// 构建运行进程
const worker = childProcess.spawn(cli, args);
// 监听控制台日志输出
worker.stdout.on("data", (message) => {
  console.log(message.toString());
  // 构造消息体
  const logLine: ILogLine = {
    date: new Date().getTime(),
    level: LogLevel.INFO,
    message: message.toString(),
  };
  /** 收集/上报日志 */
  this.logClient.pushLog(logLine);
});

// 监听控制台错误日志输出
worker.stderr.on("data", (message) => {
  console.error(message.toString());
  // 构造消息体
  const logLine: ILogLine = {
    date: new Date().getTime(),
    level: LogLevel.ERROR,
    message: message.toString(),
  };
  /** 收集/上报日志 */
  this.logClient.pushLog(logLine);
});

// 构建结束
worker.on("exit", () => {
  this.logClient.complete();
});

import { ILog, ILogLine, ILogReportConfig } from "../type/log";
import { apis } from "../api";
import * as rx from "rxjs";
import * as rxOp from "rxjs/operators";
import { LogLevel } from "../const";

/**
 * 日志上报
 */
export class LogReportClient {
  private readonly _config: ILogReportConfig;
  private _queue: rx.Subject<ILogLine>;
  private _fileBoxClient: FileBoxClient;
  private _logs: ILogLine[] = [];

  public constructor(config: ILogReportConfig) {
    this._config = config;
    this._queue = new rx.Subject<ILogLine>();
    this._initSubscribe();
  }

  /**
   * 初始化订阅器处理
   * @private
   */
  private _initSubscribe(): void {
    this._queue
      .pipe(
        rxOp.bufferCount(this._config.limit),
        rxOp.tap(async (value) => {
					this._logs.push(...value);
          // 上报日志
          await this._report(value);
        }),
      )
      .subscribe({
        complete: async () => {
          // 保证最后一次上报是最后发的
          setTimeout(() => {
            // 发送isEnd日志
            const endMessage: ILogLine = {
              message: "构建结束",
              date: new Date().getTime(),
              level: LogLevel.INFO,
            };
            void this._report([endMessage], true);
          }, 300);

          // 上报到OSS
          const buffer = new Buffer(JSON.stringify(this._logs));
          await this.ossClient.uploadBuildLog(buffer);
        },
      });
  }

  /**
   * 收集日志
   * @param log
   */
  public pushLog(log: ILogLine): void {
    this._queue.next(log);
  }

  /**
   * 构建结束
   */
  public complete(): void {
    this._queue.complete();
  }

  /**
   * 日志上报
   */
  private async _report(logs: ILogLine[], isEnd?: boolean): Promise<void> {
    const content: ILog = {
      logs: logs,
      isEnd: !!isEnd,
    };
    return apis.log
      .report(content)
      .catch((e) => {
        console.error("日志推送失败", e);
      });
  }
}

• 用 next 方法 进行日志的 push

• rxOp.bufferCount 用来堵塞流程，会在日志达到指定条数后才执行后续流程

• 再次通过 rxOp.tap 拿到分组日志进行收集和上报

• 日志收集结束后，发送结束日志并把收集到的所有日志上传到阿里云