消息队列与监控

本文包含三个主题：为什么前端监控系统要用 RabbitMQ（消息队列）、如何记录请求日志、如何拿到服务器 CPU 内存磁盘状态。

为什么前端监控系统要用 RabbitMQ？

前端监控系统是采集用户端的异常、性能、业务埋点等数据上报，在服务端做存储，并支持可视化分析的平台。

用户量可能很大，采集的数据可能比较多，这时候服务端的并发压力会比较大，要是直接存入数据库，那数据库服务很可能会崩掉。那如何保证面对大量并发请求的时候，服务不崩呢？答案就是消息队列，比如常用的 RabbitMQ：

第一个 web 服务接收请求，把消息存入 RabbitMQ，然后另一个 web 服务从 MQ 中取出消息存入数据库。

MQ 的并发量比数据库高很多。之前 web 服务要等数据库存储完成才能响应，而现在只存入 MQ 就可以响应了，那可以支持的并发量就更多。而数据库的并发比较低，可以通过 MQ 把消费的上限调低，就能保证数据库服务不崩。

比如 10w 的消息进来，每次只从中取出 1000 来消费，并发量被控制住了，自然就崩不了了。这就是 MQ 的流量削峰的功能。

而且完全可以加几个 web 服务来同时消费 MQ 中的消息（应用解耦）。

RabbitMQ 快速上手

通过 docker 来跑 RabbitMQ，搜索 rabbitmq 的镜像，选择 4.0-management 的版本（这个版本是有 web 管理界面的），点击 run，映射容器内的 5672、15672 这俩端口到本地端口。15672 是管理界面的，5672 是 mq 服务的端口。

等 rabbitmq 跑起来之后，在浏览器访问 http://localhost:15672，输入 guest、guest 进入管理页面，可以看到 connection、channel、exchange、queue 的分别的管理页面。

写个 demo 就理解了：

code

mkdir rabbitmq-test
cd rabbitmq-test
npm init -y
npm install amqplib

创建 src/producer.js：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertQueue('aaa');
await channel.sendToQueue('aaa',Buffer.from('hello'))

安装 amqplib 的包，这个是 rabbitmq 的 node 客户端（amqp 是 rabbitmq 的协议）。上面的代码连接了 rabbitmq 服务，创建了一个名字为 aaa 的队列，并向队列中发送了一个消息。（这里要用 es module 语法并且支持顶层 await 需要在 package.json 里设置 type 为 module。）

然后再写一个消费端 src/consumer.js：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

const { queue } = await channel.assertQueue('aaa');
channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

assertQueue 是如果没有就创建队列，有的话就直接返回。这样，就完成了第一次 RabbitMQ 的通信，两个服务之间也是这样通信的。

流量削峰的例子

那怎么控制并发数呢？改一下 src/producer.js，生产者每 0.5s 发送一次消息：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertQueue('aaa', {durable: false});

let i = 1;
setInterval(async () => {
    const msg = 'hello' + i;
    console.log('发送消息：', msg);
    await channel.sendToQueue('aaa',Buffer.from(msg))
    i++;
}, 500);

消费者每 1s 处理一条消息：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

const { queue } = await channel.assertQueue('aaa');
channel.prefetch(3);

const currentTask = [];
channel.consume(queue, msg => {
    currentTask.push(msg);
    console.log('收到消息：', msg.content.toString());
}, { noAck: false });

setInterval(() => {
    const curMsg = currentTask.pop();
    channel.ack(curMsg);
}, 1000);

每条消息要确认之后才会在 MQ 里删除，可以收到消息自动确认，也可以手动确认。这里我把 noAck 设置为 false，也就是不自动确认，把收到的消息放入一个数组中每 1s 确认一次。然后设置 prefetch 为 3，也就是每次最多取回 3 条消息来处理。

跑一下：生产者每 0.5s 往队列里放一条消息，消费者一开始取出 3 条，然后每处理完一条取一条，保证最多并发处理 3 条。这就是流量削峰的功能。不同服务之间的速度差异可以通过 MQ 来缓冲。

RabbitMQ 整体架构

Producer 和 Consumer 分别是生产者和消费者。Connection 是连接，但不会每用一次 rabbitmq 就创建一个单独的 Connection，而是在一个 Connection 里做一下划分，叫做 Channel，每个 Channel 做自己的事情。Queue 就是两端存取消息的地方了。整个接收消息和转发消息的服务就叫做 Broker。

至于 Exchange，前面的例子没有用到，这个是把消息放到不同的队列里用的，叫做交换机。前面生产者和消费者都是直接指定了从哪个队列存取消息，那如果是一对多的场景呢？总不能一个个地调用 sendQueue 发消息吧？这时候就要找一个 Exchange（交换机）来帮完成把消息按照规则放入不同的 Queue 的工作了。

Exchange 主要有 4 种：

fanout：把消息放到这个交换机的所有 Queue
direct：把消息放到交换机的指定 key 的队列
topic：把消息放到交换机的指定 key 的队列，支持模糊匹配
headers：把消息放到交换机的满足某些 header 的队列

direct

生产者端 src/direct.js：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange', 'direct');

channel.publish('direct-test-exchange', 'aaa',  Buffer.from('hello1'));
channel.publish('direct-test-exchange', 'bbb',  Buffer.from('hello2'));
channel.publish('direct-test-exchange', 'ccc',  Buffer.from('hello3'));

不再直接 sendToQueue，而是创建一个 exchange 然后调用 publish 往这个 exchange 发消息，其中第二个参数是 routing key，也就是消息路由到哪个队列。

然后创建两个消费者：

src/direct-consumer1.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

const { queue } = await channel.assertQueue('queue1');
await channel.bindQueue(queue,  'direct-test-exchange', 'aaa');

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

src/direct-consumer2.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

const { queue } = await channel.assertQueue('queue2');
await channel.bindQueue(queue,  'direct-test-exchange', 'bbb');

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

分别创建 queue1 和 queue2 两个队列，绑定到前面创建的 direct-test-exchange 这个交换机上，指定了路由 key 分别是 aaa 和 bbb。把生产者和两个消费者跑起来，就可以看到队列 queue1 和 queue2 分别接收到了对应的消息。

code

node src/direct.js
node src/direct-consumer1.js
node src/direct-consumer2.js

topic

src/topic.js（创建 topic 类型的 Exchange，发三条消息）：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange2', 'topic');

channel.publish('direct-test-exchange2', 'aaa.1',  Buffer.from('hello1'));
channel.publish('direct-test-exchange2', 'aaa.2',  Buffer.from('hello2'));
channel.publish('direct-test-exchange2', 'bbb.1',  Buffer.from('hello3'));

两个消费者端分别创建 queue1 和 queue2，绑定到 direct-test-exchange2 的交换机下，指定路由 key 分别为 aaa.* 和 bbb.*，这里的 * 是模糊匹配的意思（消费者端也 assertExchange，如果不存在就创建，保证 exchange 一定存在）：

src/topic-consumer1.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange2', 'topic');

const { queue } = await channel.assertQueue('queue1');
await channel.bindQueue(queue,  'direct-test-exchange2', 'aaa.*');

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

src/topic-consumer2.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange2', 'topic');

const { queue } = await channel.assertQueue('queue2');
await channel.bindQueue(queue,  'direct-test-exchange2', 'bbb.*');

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

code

node src/topic.js
node src/topic-consumer1.js
node src/topic-consumer2.js

可以看到，两个消费者分别收到了不同 routing key 对应的消息。这就是 topic 类型的交换机，可以根据模糊匹配 routing key 来发消息到不同队列。

fanout

src/fanout.js（fanout 是广播消息到 Exchange 下的所有队列，不需要指定 routing key，就算指定了也会忽略）：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange3', 'fanout');

channel.publish('direct-test-exchange3', '',  Buffer.from('hello1'));
channel.publish('direct-test-exchange3', '',  Buffer.from('hello2'));
channel.publish('direct-test-exchange3', '',  Buffer.from('hello3'));

code

node src/fanout.js
node src/fanout-consumer1.js
node src/fanout-consumer2.js

跑起来可以看到两个消费者都收到了消息。这就是 fanout 类型交换机的特点，广播消息到所有绑定到它的 queue。

headers

headers 类型的 Exchange 不是根据 routing key 来匹配了，而是根据 headers。生产者端 src/headers.js：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange4', 'headers');

channel.publish('direct-test-exchange4', '',  Buffer.from('hello1'), {
    headers: {
        name: 'guang'
    }
});
channel.publish('direct-test-exchange4', '',  Buffer.from('hello2'), {
    headers: {
        name: 'guang'
    }
});
channel.publish('direct-test-exchange4', '',  Buffer.from('hello3'), {
    headers: {
        name: 'dong'
    }
});

消费者端：

src/headers-consumer1.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange4', 'headers');

const { queue } = await channel.assertQueue('queue1');
await channel.bindQueue(queue,  'direct-test-exchange4', '', {
    name: 'guang'
});

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

src/headers-consumer2.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange4', 'headers');

const { queue } = await channel.assertQueue('queue2');
await channel.bindQueue(queue,  'direct-test-exchange4', '', {
    name: 'dong'
});

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

跑起来：

code

node src/headers.js
node src/headers-consumer1.js
node src/headers-consumer2.js

这就是 Exchange，当你需要一对多发消息的时候，就可以选择这些类型的交换机。

回过头来，总结下 rabbitmq 解决了什么问题：

流量削峰：可以把很大的流量放到 mq 中按照一定的流量上限来慢慢消费，这样虽然慢一点，但不至于崩溃。
应用解耦：应用之间不再直接依赖，就算某个应用挂掉了，也可以再恢复后继续从 mq 中消费消息，并不会一个应用挂掉了，它关联的应用也挂掉。

比如前端监控系统的后端服务，就很适合使用 mq 来做流量削峰。

如何记录请求日志

Nest 服务会不断处理用户的请求，如果想记录下每次请求的日志呢？可以通过 interceptor 来做。

code

nest new request-log
nest g interceptor request-log --no-spec --flat

打印下日志：

javascript

import { CallHandler, ExecutionContext, Injectable, Logger, NestInterceptor } from '@nestjs/common';
import { Response } from 'express';
import { Request } from 'express';
import { Observable, tap } from 'rxjs';

@Injectable()
export class RequestLogInterceptor implements NestInterceptor {
  private readonly logger = new Logger(RequestLogInterceptor.name);

  intercept(
    context: ExecutionContext,
    next: CallHandler<any>,
  ) {
    const request = context.switchToHttp().getRequest<Request>();
    const response = context.switchToHttp().getResponse<Response>();

    const userAgent = request.headers['user-agent'];

    const { ip, method, path } = request;

    this.logger.debug(
      `${method} ${path} ${ip} ${userAgent}: ${
        context.getClass().name
      } ${
        context.getHandler().name
      } invoked...`,
    );

    const now = Date.now();

    return next.handle().pipe(
      tap((res) => {
        this.logger.debug(
          `${method} ${path} ${ip} ${userAgent}: ${response.statusCode}: ${Date.now() - now}ms`,
        );
        this.logger.debug(`Response: ${JSON.stringify(res)}`);
      }),
    );
  }
}

这里用 nest 的 Logger 打印日志，打印 method、path、ip、user agent、调用的目标 class、handler 等信息，然后记录下响应的状态码和请求时间还有响应内容。

全局启用这个 interceptor：

javascript

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { APP_INTERCEPTOR } from '@nestjs/core';
import { RequestLogInterceptor } from './request-log.interceptor';

@Module({
  imports: [],
  controllers: [AppController],
  providers: [
    AppService,
    {
      provide: APP_INTERCEPTOR,
      useClass: RequestLogInterceptor
    }
  ],
})
export class AppModule {}

把服务跑起来，浏览器访问下，可以看到打印了请求的信息、目标 class、handler、响应的内容。

但其实这个 ip 是有问题的：如果客户端直接请求 Nest 服务，那这个 ip 是准的，但如果中间经过了 11-Nginx基础概述等服务器的转发，那拿到的 ip 就是 11-Nginx基础概述服务器的 ip 了。这时候要取 X-Forwarded-For 这个 header，它记录着转发的客户端 ip。当然这种事情不用自己做，有专门的库 request-ip：

code

npm install --save request-ip

然后把打印的 ip 换一下（换成 X-Forwarded-For 的客户端 ip 或者是 request.ip）：

javascript

import { CallHandler, ExecutionContext, Injectable, Logger, NestInterceptor } from '@nestjs/common';
import { Response } from 'express';
import { Request } from 'express';
import { Observable, tap } from 'rxjs';
import * as requestIp from 'request-ip';

@Injectable()
export class RequestLogInterceptor implements NestInterceptor {
  private readonly logger = new Logger(RequestLogInterceptor.name);

  intercept(
    context: ExecutionContext,
    next: CallHandler<any>,
  ) {
    const request = context.switchToHttp().getRequest<Request>();
    const response = context.switchToHttp().getResponse<Response>();

    const userAgent = request.headers['user-agent'];

    const { ip, method, path } = request;

    const clientIp = requestIp.getClientIp(ip) || ip;

    this.logger.debug(
      `${method} ${path} ${clientIp} ${userAgent}: ${
        context.getClass().name
      } ${
        context.getHandler().name
      } invoked...`,
    );

    const now = Date.now();

    return next.handle().pipe(
      tap((res) => {
        this.logger.debug(
          `${method} ${path} ${clientIp} ${userAgent}: ${response.statusCode}: ${Date.now() - now}ms`,
        );
        this.logger.debug(`Response: ${JSON.stringify(res)}`);
      }),
    );
  }
}

因为本地访问用 localhost 是拿不到真实 ip 的，你可以查一下本地 ip，用 ip 访问，这里的 ::ffff 是 ipv6 地址的意思，这样部署到线上之后就能拿到真实地址了。

根据 ip 拿到城市信息

很多系统会做登录日志，每次登录的时候记录登录时的 ip 和对应的城市信息到数据库里。如何根据 ip 拿到城市信息呢？其实可以通过一些在线的免费接口：

https://whois.pconline.com.cn/ipJson.jsp?ip=221.237.121.165&json=true

请求三方服务用 axios 的包，安装下：

code

npm install --save @nestjs/axios axios

在 AppModule 里引入下，然后在 interceptor 里注入 HttpService 来发请求，封装个 ipToCity 方法：

javascript

import { CallHandler, ExecutionContext, Inject, Injectable, Logger, NestInterceptor } from '@nestjs/common';
import { Response } from 'express';
import { Request } from 'express';
import { Observable, tap } from 'rxjs';
import * as requestIp from 'request-ip';
import { HttpService } from '@nestjs/axios';

@Injectable()
export class RequestLogInterceptor implements NestInterceptor {
  private readonly logger = new Logger(RequestLogInterceptor.name);

  @Inject(HttpService)
  private httpService: HttpService;

  async ipToCity(ip: string) {
    const response = await this.httpService.axiosRef(`https://whois.pconline.com.cn/ipJson.jsp?ip=${ip}&json=true`);
    return response.data.addr;
  }

  async intercept(
    context: ExecutionContext,
    next: CallHandler<any>,
  ) {
    const request = context.switchToHttp().getRequest<Request>();
    const response = context.switchToHttp().getResponse<Response>();

    console.log(await this.ipToCity('221.237.121.165'))

    const userAgent = request.headers['user-agent'];

    const { ip, method, path } = request;

    const clientIp = requestIp.getClientIp(ip) || ip;

    this.logger.debug(
      `${method} ${path} ${clientIp} ${userAgent}: ${
        context.getClass().name
      } ${
        context.getHandler().name
      } invoked...`,
    );

    const now = Date.now();

    return next.handle().pipe(
      tap((res) => {
        this.logger.debug(
          `${method} ${path} ${clientIp} ${userAgent}: ${response.statusCode}: ${Date.now() - now}ms`,
        );
        this.logger.debug(`Response: ${JSON.stringify(res)}`);
      }),
    );
  }
}

直接用 httpService 的方法是被包装过后的，返回值是 rxjs 的 Observable，需要用 firstValueFrom 转为 promise。如果想用原生 axios 对象，可以直接调用 this.httpService.axiosRef.xxx，这样返回的就是 promise。

接口返回的字符集是 gbk，而用的是 utf-8，所以需要转换一下，用 iconv-lite 这个包：

code

npm install --save iconv-lite

指定 responseType 为 arraybuffer，也就是二进制的数组，然后用 gbk 的字符集来解码：

javascript

import * as iconv from 'iconv-lite';

async ipToCity(ip: string) {
    const response = await this.httpService.axiosRef(`https://whois.pconline.com.cn/ipJson.jsp?ip=${ip}&json=true`, {
      responseType: 'arraybuffer',
      transformResponse: [
        function (data) {
          const str = iconv.decode(data, 'gbk');
          return JSON.parse(str);
        }
      ]
    });
    return response.data.addr;
}

现在就能拿到 utf-8 编码的城市信息了。当然，这个不建议放到请求日志里，不然每次请求都调用一次接口太浪费性能了，登录日志里可以加这个。

如何拿到服务器 CPU、内存、磁盘状态

在很多系统中，会有展示服务器信息的页面，比如 CPU、内存、磁盘等。这是通过 node 的原生 api 就可以做到的。

还是用 Nest 写个接口来返回这些信息：

code

nest new server-status

在 AppController 加个接口，用 os.cpus 拿到 cpu 的信息：

javascript

import { Controller, Get } from '@nestjs/common';
import { AppService } from './app.service';
import * as os from 'os';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get()
  getHello(): string {
    return this.appService.getHello();
  }

  @Get('status')
  status() {
    return os.cpus();
  }
}

把服务跑起来，返回的数组元素个数就是 cpu 数。可以看到 times.user、times.sys、times.idle 分别代表用户代码占用的 cpu 时间、系统代码占用的 cpu 时间、空闲的 cpu 时间。基于这些就能算出 cpu 的使用率、空置率来。

CPU 使用率计算

javascript

@Get('status')
status() {
    const cpus = os.cpus();
    const cpuInfo = cpus.reduce(
      (info, cpu) => {
        info.cpuNum += 1;
        info.user += cpu.times.user;
        info.sys += cpu.times.sys;
        info.idle += cpu.times.idle;
        info.total += cpu.times.user + cpu.times.sys + cpu.times.idle;
        return info;
      },
      { user: 0, sys: 0, idle: 0, total: 0, cpuNum: 0 },
    );
    const cpu = {
      cpuNum: cpuInfo.cpuNum,
      sys: ((cpuInfo.sys / cpuInfo.total) * 100).toFixed(2),
      used: ((cpuInfo.user / cpuInfo.total) * 100).toFixed(2),
      free: ((cpuInfo.idle / cpuInfo.total) * 100).toFixed(2),
    };
    return cpu;
}

用 reduce 方法累加 cpu 的数量、user、sys、idle 以及总的 cpu 时间，然后 cpu 的系统使用率就是 sys/total，用户使用率是 user/total，而空置率就是 idle/total。

内存信息

继续拿内存信息，把 CPU 和内存整合方便统一返回：

javascript

import { Controller, Get } from '@nestjs/common';
import { AppService } from './app.service';
import * as os from 'os';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get()
  getHello(): string {
    return this.appService.getHello();
  }

  @Get('status')
  status() {
    return {
      cpu: this.getCpuInfo(),
      mem: this.getMemInfo()
    }
  }

  getMemInfo() {
    const totalMemory = os.totalmem();
    const freeMemory = os.freemem();
    const usedMemory = totalMemory - freeMemory;
    const memoryUsagePercentage = (((totalMemory - freeMemory) / totalMemory) * 100).toFixed(2);
    const mem = {
      total: totalMemory,
      used: usedMemory,
      free: freeMemory,
      usage: memoryUsagePercentage,
    };
    return mem;
  }

  getCpuInfo() {
    const cpus = os.cpus();
    const cpuInfo = cpus.reduce(
      (info, cpu) => {
        info.cpuNum += 1;
        info.user += cpu.times.user;
        info.sys += cpu.times.sys;
        info.idle += cpu.times.idle;
        info.total += cpu.times.user + cpu.times.sys + cpu.times.idle;
        return info;
      },
      { user: 0, sys: 0, idle: 0, total: 0, cpuNum: 0 },
    );
    const cpu = {
      cpuNum: cpuInfo.cpuNum,
      sys: ((cpuInfo.sys / cpuInfo.total) * 100).toFixed(2),
      used: ((cpuInfo.user / cpuInfo.total) * 100).toFixed(2),
      free: ((cpuInfo.idle / cpuInfo.total) * 100).toFixed(2),
    };
    return cpu;
  }
}

os.totalmem 是总内存，os.freemem 是空闲内存，那差值就是已使用的内存，这样也可以求出内存使用率。可以把它转成 GB 单位的数值返回：

javascript

bytesToGB(bytes) {
    const gb = bytes / (1024 * 1024 * 1024);
    return gb.toFixed(2);
}

getMemInfo() {
    const totalMemory = os.totalmem();
    const freeMemory = os.freemem();
    const usedMemory = totalMemory - freeMemory;
    const memoryUsagePercentage = (((totalMemory - freeMemory) / totalMemory) * 100).toFixed(2);
    const mem = {
      total: this.bytesToGB(totalMemory),
      used: this.bytesToGB(usedMemory),
      free: this.bytesToGB(freeMemory),
      usage: memoryUsagePercentage,
    };
    return mem;
}

磁盘信息

然后是磁盘信息，这里用到 node-disk-info 这个包：

code

npm install --save node-disk-info

调用下：

javascript

import { Controller, Get } from '@nestjs/common';
import { AppService } from './app.service';
import * as os from 'os';
import * as nodeDiskInfo from 'node-disk-info';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get()
  getHello(): string {
    return this.appService.getHello();
  }

  @Get('status')
  async status() {
    return {
      cpu: this.getCpuInfo(),
      mem: this.getMemInfo(),
      dist: await this.getDiskStatus()
    }
  }

  async getDiskStatus() {
    const disks = await nodeDiskInfo.getDiskInfoSync();

    const sysFiles = disks.map((disk: any) => {
      return {
        dirName: disk._mounted,
        typeName: disk._filesystem,
        total: this.bytesToGB(disk._blocks) + 'GB',
        used: this.bytesToGB(disk._used) + 'GB',
        free: this.bytesToGB(disk._available) + 'GB',
        usage: ((disk._used / disk._blocks || 0) * 100).toFixed(2),
      };
    });
    return sysFiles;
  }

  bytesToGB(bytes) {
    const gb = bytes / (1024 * 1024 * 1024);
    return gb.toFixed(2);
  }

  getMemInfo() {
    const totalMemory = os.totalmem();
    const freeMemory = os.freemem();
    const usedMemory = totalMemory - freeMemory;
    const memoryUsagePercentage = (((totalMemory - freeMemory) / totalMemory) * 100).toFixed(2);
    const mem = {
      total: this.bytesToGB(totalMemory),
      used: this.bytesToGB(usedMemory),
      free: this.bytesToGB(freeMemory),
      usage: memoryUsagePercentage
    };
    return mem;
  }

  getCpuInfo() {
    const cpus = os.cpus();
    const cpuInfo = cpus.reduce(
      (info, cpu) => {
        info.cpuNum += 1;
        info.user += cpu.times.user;
        info.sys += cpu.times.sys;
        info.idle += cpu.times.idle;
        info.total += cpu.times.user + cpu.times.sys + cpu.times.idle;
        return info;
      },
      { user: 0, sys: 0, idle: 0, total: 0, cpuNum: 0 },
    );
    const cpu = {
      cpuNum: cpuInfo.cpuNum,
      sys: ((cpuInfo.sys / cpuInfo.total) * 100).toFixed(2),
      used: ((cpuInfo.user / cpuInfo.total) * 100).toFixed(2),
      free: ((cpuInfo.idle / cpuInfo.total) * 100).toFixed(2),
    };
    return cpu;
  }
}

这里拿到的就是本地所有的磁盘信息，分别是路径、文件系统、总大小、已用大小、可用大小、已用百分比。

其他服务器信息

最后，还要返回其他的服务器信息，如主机名、IP、操作系统等：

javascript

@Get('status')
async status() {
    return {
      cpu: this.getCpuInfo(),
      mem: this.getMemInfo(),
      dist: await this.getDiskStatus(),
      sys: this.getSysInfo()
    }
}

getSysInfo() {
    return {
      computerName: os.hostname(),
      computerIp: this.getServerIP(),
      osName: os.platform(),
      osArch: os.arch(),
    };
}

getServerIP() {
    const nets = os.networkInterfaces();
    for (const name of Object.keys(nets)) {
      for (const net of nets[name]) {
        if (net.family === 'IPv4' && !net.internal) {
          return net.address;
        }
      }
    }
}

这里的 os.networkInterfaces 是拿到所有网卡信息，从中过滤出非内部的 IPv4 外部网卡的 ip 来返回。此外，还通过 os.hostname、os.platform、os.arch 分别拿到了主机名、操作系统、操作系统架构等信息。

试一下，和我系统设置里的 ip 一样。这样，服务器的状态数据就都拿到了，可以实现这样的服务监控页面。

总结

RabbitMQ：前端监控系统会收到很多来自用户端的请求，如果直接存入数据库很容易把数据库服务搞挂掉，所以一般会加一个 RabbitMQ 来缓冲。它是生产者往 queue 里放入消息，消费者从里面读消息，之后确认消息收到的流程。当一对多的时候，还要加一个 Exchange 交换机来根据不同的规则转发消息（direct/topic/headers/fanout）。消费者可以设置一个消费的并发上限，保证服务不会因并发过高而崩溃，这就是流量削峰。
请求日志：通过 interceptor 实现了记录请求日志的功能。其中 ip 地址如果被 11-Nginx基础概述转发过，需要取 X-Forwarded-For 的 header 的值，直接用 request-ip 这个包来做。如果想拿到 ip 对应的城市信息，可以用一些免费接口来查询，用 @nestjs/axios 来发送请求（不建议放到请求日志里）。
服务器状态：拿到了服务器的 cpu、内存、磁盘、ip 等信息，通过 node 的 os 模块的 api 以及 node-disk-info 这个包。可以在后台管理系统加一个页面来展示这些信息。

消息队列与监控

本文包含三个主题：为什么前端监控系统要用 RabbitMQ（消息队列）、如何记录请求日志、如何拿到服务器 CPU 内存磁盘状态。

为什么前端监控系统要用 RabbitMQ？

前端监控系统是采集用户端的异常、性能、业务埋点等数据上报，在服务端做存储，并支持可视化分析的平台。

第一个 web 服务接收请求，把消息存入 RabbitMQ，然后另一个 web 服务从 MQ 中取出消息存入数据库。

比如 10w 的消息进来，每次只从中取出 1000 来消费，并发量被控制住了，自然就崩不了了。这就是 MQ 的流量削峰的功能。

而且完全可以加几个 web 服务来同时消费 MQ 中的消息（应用解耦）。

RabbitMQ 快速上手

等 rabbitmq 跑起来之后，在浏览器访问 http://localhost:15672，输入 guest、guest 进入管理页面，可以看到 connection、channel、exchange、queue 的分别的管理页面。

写个 demo 就理解了：

code

mkdir rabbitmq-test
cd rabbitmq-test
npm init -y
npm install amqplib

创建 src/producer.js：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertQueue('aaa');
await channel.sendToQueue('aaa',Buffer.from('hello'))

然后再写一个消费端 src/consumer.js：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

const { queue } = await channel.assertQueue('aaa');
channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

assertQueue 是如果没有就创建队列，有的话就直接返回。这样，就完成了第一次 RabbitMQ 的通信，两个服务之间也是这样通信的。

流量削峰的例子

那怎么控制并发数呢？改一下 src/producer.js，生产者每 0.5s 发送一次消息：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertQueue('aaa', {durable: false});

let i = 1;
setInterval(async () => {
    const msg = 'hello' + i;
    console.log('发送消息：', msg);
    await channel.sendToQueue('aaa',Buffer.from(msg))
    i++;
}, 500);

消费者每 1s 处理一条消息：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

const { queue } = await channel.assertQueue('aaa');
channel.prefetch(3);

const currentTask = [];
channel.consume(queue, msg => {
    currentTask.push(msg);
    console.log('收到消息：', msg.content.toString());
}, { noAck: false });

setInterval(() => {
    const curMsg = currentTask.pop();
    channel.ack(curMsg);
}, 1000);

RabbitMQ 整体架构

Exchange 主要有 4 种：

fanout：把消息放到这个交换机的所有 Queue
direct：把消息放到交换机的指定 key 的队列
topic：把消息放到交换机的指定 key 的队列，支持模糊匹配
headers：把消息放到交换机的满足某些 header 的队列

direct

生产者端 src/direct.js：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange', 'direct');

channel.publish('direct-test-exchange', 'aaa',  Buffer.from('hello1'));
channel.publish('direct-test-exchange', 'bbb',  Buffer.from('hello2'));
channel.publish('direct-test-exchange', 'ccc',  Buffer.from('hello3'));

不再直接 sendToQueue，而是创建一个 exchange 然后调用 publish 往这个 exchange 发消息，其中第二个参数是 routing key，也就是消息路由到哪个队列。

然后创建两个消费者：

src/direct-consumer1.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

const { queue } = await channel.assertQueue('queue1');
await channel.bindQueue(queue,  'direct-test-exchange', 'aaa');

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

src/direct-consumer2.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

const { queue } = await channel.assertQueue('queue2');
await channel.bindQueue(queue,  'direct-test-exchange', 'bbb');

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

code

node src/direct.js
node src/direct-consumer1.js
node src/direct-consumer2.js

topic

src/topic.js（创建 topic 类型的 Exchange，发三条消息）：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange2', 'topic');

channel.publish('direct-test-exchange2', 'aaa.1',  Buffer.from('hello1'));
channel.publish('direct-test-exchange2', 'aaa.2',  Buffer.from('hello2'));
channel.publish('direct-test-exchange2', 'bbb.1',  Buffer.from('hello3'));

src/topic-consumer1.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange2', 'topic');

const { queue } = await channel.assertQueue('queue1');
await channel.bindQueue(queue,  'direct-test-exchange2', 'aaa.*');

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

src/topic-consumer2.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange2', 'topic');

const { queue } = await channel.assertQueue('queue2');
await channel.bindQueue(queue,  'direct-test-exchange2', 'bbb.*');

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

code

node src/topic.js
node src/topic-consumer1.js
node src/topic-consumer2.js

可以看到，两个消费者分别收到了不同 routing key 对应的消息。这就是 topic 类型的交换机，可以根据模糊匹配 routing key 来发消息到不同队列。

fanout

src/fanout.js（fanout 是广播消息到 Exchange 下的所有队列，不需要指定 routing key，就算指定了也会忽略）：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange3', 'fanout');

channel.publish('direct-test-exchange3', '',  Buffer.from('hello1'));
channel.publish('direct-test-exchange3', '',  Buffer.from('hello2'));
channel.publish('direct-test-exchange3', '',  Buffer.from('hello3'));

code

node src/fanout.js
node src/fanout-consumer1.js
node src/fanout-consumer2.js

跑起来可以看到两个消费者都收到了消息。这就是 fanout 类型交换机的特点，广播消息到所有绑定到它的 queue。

headers

headers 类型的 Exchange 不是根据 routing key 来匹配了，而是根据 headers。生产者端 src/headers.js：

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange4', 'headers');

channel.publish('direct-test-exchange4', '',  Buffer.from('hello1'), {
    headers: {
        name: 'guang'
    }
});
channel.publish('direct-test-exchange4', '',  Buffer.from('hello2'), {
    headers: {
        name: 'guang'
    }
});
channel.publish('direct-test-exchange4', '',  Buffer.from('hello3'), {
    headers: {
        name: 'dong'
    }
});

消费者端：

src/headers-consumer1.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange4', 'headers');

const { queue } = await channel.assertQueue('queue1');
await channel.bindQueue(queue,  'direct-test-exchange4', '', {
    name: 'guang'
});

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

src/headers-consumer2.js

javascript

import * as amqp from 'amqplib'

const connect = await amqp.connect(`amqp://localhost:5672`);
const channel = await connect.createChannel();

await channel.assertExchange('direct-test-exchange4', 'headers');

const { queue } = await channel.assertQueue('queue2');
await channel.bindQueue(queue,  'direct-test-exchange4', '', {
    name: 'dong'
});

channel.consume(queue, msg => {
    console.log(msg.content.toString())
}, { noAck: true });

跑起来：

code

node src/headers.js
node src/headers-consumer1.js
node src/headers-consumer2.js

这就是 Exchange，当你需要一对多发消息的时候，就可以选择这些类型的交换机。

回过头来，总结下 rabbitmq 解决了什么问题：

流量削峰：可以把很大的流量放到 mq 中按照一定的流量上限来慢慢消费，这样虽然慢一点，但不至于崩溃。
应用解耦：应用之间不再直接依赖，就算某个应用挂掉了，也可以再恢复后继续从 mq 中消费消息，并不会一个应用挂掉了，它关联的应用也挂掉。

比如前端监控系统的后端服务，就很适合使用 mq 来做流量削峰。

如何记录请求日志

Nest 服务会不断处理用户的请求，如果想记录下每次请求的日志呢？可以通过 interceptor 来做。

code

nest new request-log
nest g interceptor request-log --no-spec --flat

打印下日志：

javascript

import { CallHandler, ExecutionContext, Injectable, Logger, NestInterceptor } from '@nestjs/common';
import { Response } from 'express';
import { Request } from 'express';
import { Observable, tap } from 'rxjs';

@Injectable()
export class RequestLogInterceptor implements NestInterceptor {
  private readonly logger = new Logger(RequestLogInterceptor.name);

  intercept(
    context: ExecutionContext,
    next: CallHandler<any>,
  ) {
    const request = context.switchToHttp().getRequest<Request>();
    const response = context.switchToHttp().getResponse<Response>();

    const userAgent = request.headers['user-agent'];

    const { ip, method, path } = request;

    this.logger.debug(
      `${method} ${path} ${ip} ${userAgent}: ${
        context.getClass().name
      } ${
        context.getHandler().name
      } invoked...`,
    );

    const now = Date.now();

    return next.handle().pipe(
      tap((res) => {
        this.logger.debug(
          `${method} ${path} ${ip} ${userAgent}: ${response.statusCode}: ${Date.now() - now}ms`,
        );
        this.logger.debug(`Response: ${JSON.stringify(res)}`);
      }),
    );
  }
}

这里用 nest 的 Logger 打印日志，打印 method、path、ip、user agent、调用的目标 class、handler 等信息，然后记录下响应的状态码和请求时间还有响应内容。

全局启用这个 interceptor：

javascript

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { APP_INTERCEPTOR } from '@nestjs/core';
import { RequestLogInterceptor } from './request-log.interceptor';

@Module({
  imports: [],
  controllers: [AppController],
  providers: [
    AppService,
    {
      provide: APP_INTERCEPTOR,
      useClass: RequestLogInterceptor
    }
  ],
})
export class AppModule {}

把服务跑起来，浏览器访问下，可以看到打印了请求的信息、目标 class、handler、响应的内容。

code

npm install --save request-ip

然后把打印的 ip 换一下（换成 X-Forwarded-For 的客户端 ip 或者是 request.ip）：

javascript

import { CallHandler, ExecutionContext, Injectable, Logger, NestInterceptor } from '@nestjs/common';
import { Response } from 'express';
import { Request } from 'express';
import { Observable, tap } from 'rxjs';
import * as requestIp from 'request-ip';

@Injectable()
export class RequestLogInterceptor implements NestInterceptor {
  private readonly logger = new Logger(RequestLogInterceptor.name);

  intercept(
    context: ExecutionContext,
    next: CallHandler<any>,
  ) {
    const request = context.switchToHttp().getRequest<Request>();
    const response = context.switchToHttp().getResponse<Response>();

    const userAgent = request.headers['user-agent'];

    const { ip, method, path } = request;

    const clientIp = requestIp.getClientIp(ip) || ip;

    this.logger.debug(
      `${method} ${path} ${clientIp} ${userAgent}: ${
        context.getClass().name
      } ${
        context.getHandler().name
      } invoked...`,
    );

    const now = Date.now();

    return next.handle().pipe(
      tap((res) => {
        this.logger.debug(
          `${method} ${path} ${clientIp} ${userAgent}: ${response.statusCode}: ${Date.now() - now}ms`,
        );
        this.logger.debug(`Response: ${JSON.stringify(res)}`);
      }),
    );
  }
}

根据 ip 拿到城市信息

https://whois.pconline.com.cn/ipJson.jsp?ip=221.237.121.165&json=true

请求三方服务用 axios 的包，安装下：

code

npm install --save @nestjs/axios axios

在 AppModule 里引入下，然后在 interceptor 里注入 HttpService 来发请求，封装个 ipToCity 方法：

javascript

import { CallHandler, ExecutionContext, Inject, Injectable, Logger, NestInterceptor } from '@nestjs/common';
import { Response } from 'express';
import { Request } from 'express';
import { Observable, tap } from 'rxjs';
import * as requestIp from 'request-ip';
import { HttpService } from '@nestjs/axios';

@Injectable()
export class RequestLogInterceptor implements NestInterceptor {
  private readonly logger = new Logger(RequestLogInterceptor.name);

  @Inject(HttpService)
  private httpService: HttpService;

  async ipToCity(ip: string) {
    const response = await this.httpService.axiosRef(`https://whois.pconline.com.cn/ipJson.jsp?ip=${ip}&json=true`);
    return response.data.addr;
  }

  async intercept(
    context: ExecutionContext,
    next: CallHandler<any>,
  ) {
    const request = context.switchToHttp().getRequest<Request>();
    const response = context.switchToHttp().getResponse<Response>();

    console.log(await this.ipToCity('221.237.121.165'))

    const userAgent = request.headers['user-agent'];

    const { ip, method, path } = request;

    const clientIp = requestIp.getClientIp(ip) || ip;

    this.logger.debug(
      `${method} ${path} ${clientIp} ${userAgent}: ${
        context.getClass().name
      } ${
        context.getHandler().name
      } invoked...`,
    );

    const now = Date.now();

    return next.handle().pipe(
      tap((res) => {
        this.logger.debug(
          `${method} ${path} ${clientIp} ${userAgent}: ${response.statusCode}: ${Date.now() - now}ms`,
        );
        this.logger.debug(`Response: ${JSON.stringify(res)}`);
      }),
    );
  }
}

接口返回的字符集是 gbk，而用的是 utf-8，所以需要转换一下，用 iconv-lite 这个包：

code

npm install --save iconv-lite

指定 responseType 为 arraybuffer，也就是二进制的数组，然后用 gbk 的字符集来解码：

javascript

import * as iconv from 'iconv-lite';

async ipToCity(ip: string) {
    const response = await this.httpService.axiosRef(`https://whois.pconline.com.cn/ipJson.jsp?ip=${ip}&json=true`, {
      responseType: 'arraybuffer',
      transformResponse: [
        function (data) {
          const str = iconv.decode(data, 'gbk');
          return JSON.parse(str);
        }
      ]
    });
    return response.data.addr;
}

现在就能拿到 utf-8 编码的城市信息了。当然，这个不建议放到请求日志里，不然每次请求都调用一次接口太浪费性能了，登录日志里可以加这个。

如何拿到服务器 CPU、内存、磁盘状态

在很多系统中，会有展示服务器信息的页面，比如 CPU、内存、磁盘等。这是通过 node 的原生 api 就可以做到的。

还是用 Nest 写个接口来返回这些信息：

code

nest new server-status

在 AppController 加个接口，用 os.cpus 拿到 cpu 的信息：

javascript

import { Controller, Get } from '@nestjs/common';
import { AppService } from './app.service';
import * as os from 'os';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get()
  getHello(): string {
    return this.appService.getHello();
  }

  @Get('status')
  status() {
    return os.cpus();
  }
}

CPU 使用率计算

javascript

@Get('status')
status() {
    const cpus = os.cpus();
    const cpuInfo = cpus.reduce(
      (info, cpu) => {
        info.cpuNum += 1;
        info.user += cpu.times.user;
        info.sys += cpu.times.sys;
        info.idle += cpu.times.idle;
        info.total += cpu.times.user + cpu.times.sys + cpu.times.idle;
        return info;
      },
      { user: 0, sys: 0, idle: 0, total: 0, cpuNum: 0 },
    );
    const cpu = {
      cpuNum: cpuInfo.cpuNum,
      sys: ((cpuInfo.sys / cpuInfo.total) * 100).toFixed(2),
      used: ((cpuInfo.user / cpuInfo.total) * 100).toFixed(2),
      free: ((cpuInfo.idle / cpuInfo.total) * 100).toFixed(2),
    };
    return cpu;
}

用 reduce 方法累加 cpu 的数量、user、sys、idle 以及总的 cpu 时间，然后 cpu 的系统使用率就是 sys/total，用户使用率是 user/total，而空置率就是 idle/total。

内存信息

继续拿内存信息，把 CPU 和内存整合方便统一返回：

javascript

import { Controller, Get } from '@nestjs/common';
import { AppService } from './app.service';
import * as os from 'os';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get()
  getHello(): string {
    return this.appService.getHello();
  }

  @Get('status')
  status() {
    return {
      cpu: this.getCpuInfo(),
      mem: this.getMemInfo()
    }
  }

  getMemInfo() {
    const totalMemory = os.totalmem();
    const freeMemory = os.freemem();
    const usedMemory = totalMemory - freeMemory;
    const memoryUsagePercentage = (((totalMemory - freeMemory) / totalMemory) * 100).toFixed(2);
    const mem = {
      total: totalMemory,
      used: usedMemory,
      free: freeMemory,
      usage: memoryUsagePercentage,
    };
    return mem;
  }

  getCpuInfo() {
    const cpus = os.cpus();
    const cpuInfo = cpus.reduce(
      (info, cpu) => {
        info.cpuNum += 1;
        info.user += cpu.times.user;
        info.sys += cpu.times.sys;
        info.idle += cpu.times.idle;
        info.total += cpu.times.user + cpu.times.sys + cpu.times.idle;
        return info;
      },
      { user: 0, sys: 0, idle: 0, total: 0, cpuNum: 0 },
    );
    const cpu = {
      cpuNum: cpuInfo.cpuNum,
      sys: ((cpuInfo.sys / cpuInfo.total) * 100).toFixed(2),
      used: ((cpuInfo.user / cpuInfo.total) * 100).toFixed(2),
      free: ((cpuInfo.idle / cpuInfo.total) * 100).toFixed(2),
    };
    return cpu;
  }
}

os.totalmem 是总内存，os.freemem 是空闲内存，那差值就是已使用的内存，这样也可以求出内存使用率。可以把它转成 GB 单位的数值返回：

javascript

bytesToGB(bytes) {
    const gb = bytes / (1024 * 1024 * 1024);
    return gb.toFixed(2);
}

getMemInfo() {
    const totalMemory = os.totalmem();
    const freeMemory = os.freemem();
    const usedMemory = totalMemory - freeMemory;
    const memoryUsagePercentage = (((totalMemory - freeMemory) / totalMemory) * 100).toFixed(2);
    const mem = {
      total: this.bytesToGB(totalMemory),
      used: this.bytesToGB(usedMemory),
      free: this.bytesToGB(freeMemory),
      usage: memoryUsagePercentage,
    };
    return mem;
}

磁盘信息

然后是磁盘信息，这里用到 node-disk-info 这个包：

code

npm install --save node-disk-info

调用下：

javascript

import { Controller, Get } from '@nestjs/common';
import { AppService } from './app.service';
import * as os from 'os';
import * as nodeDiskInfo from 'node-disk-info';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get()
  getHello(): string {
    return this.appService.getHello();
  }

  @Get('status')
  async status() {
    return {
      cpu: this.getCpuInfo(),
      mem: this.getMemInfo(),
      dist: await this.getDiskStatus()
    }
  }

  async getDiskStatus() {
    const disks = await nodeDiskInfo.getDiskInfoSync();

    const sysFiles = disks.map((disk: any) => {
      return {
        dirName: disk._mounted,
        typeName: disk._filesystem,
        total: this.bytesToGB(disk._blocks) + 'GB',
        used: this.bytesToGB(disk._used) + 'GB',
        free: this.bytesToGB(disk._available) + 'GB',
        usage: ((disk._used / disk._blocks || 0) * 100).toFixed(2),
      };
    });
    return sysFiles;
  }

  bytesToGB(bytes) {
    const gb = bytes / (1024 * 1024 * 1024);
    return gb.toFixed(2);
  }

  getMemInfo() {
    const totalMemory = os.totalmem();
    const freeMemory = os.freemem();
    const usedMemory = totalMemory - freeMemory;
    const memoryUsagePercentage = (((totalMemory - freeMemory) / totalMemory) * 100).toFixed(2);
    const mem = {
      total: this.bytesToGB(totalMemory),
      used: this.bytesToGB(usedMemory),
      free: this.bytesToGB(freeMemory),
      usage: memoryUsagePercentage
    };
    return mem;
  }

  getCpuInfo() {
    const cpus = os.cpus();
    const cpuInfo = cpus.reduce(
      (info, cpu) => {
        info.cpuNum += 1;
        info.user += cpu.times.user;
        info.sys += cpu.times.sys;
        info.idle += cpu.times.idle;
        info.total += cpu.times.user + cpu.times.sys + cpu.times.idle;
        return info;
      },
      { user: 0, sys: 0, idle: 0, total: 0, cpuNum: 0 },
    );
    const cpu = {
      cpuNum: cpuInfo.cpuNum,
      sys: ((cpuInfo.sys / cpuInfo.total) * 100).toFixed(2),
      used: ((cpuInfo.user / cpuInfo.total) * 100).toFixed(2),
      free: ((cpuInfo.idle / cpuInfo.total) * 100).toFixed(2),
    };
    return cpu;
  }
}

这里拿到的就是本地所有的磁盘信息，分别是路径、文件系统、总大小、已用大小、可用大小、已用百分比。

其他服务器信息

最后，还要返回其他的服务器信息，如主机名、IP、操作系统等：

javascript

@Get('status')
async status() {
    return {
      cpu: this.getCpuInfo(),
      mem: this.getMemInfo(),
      dist: await this.getDiskStatus(),
      sys: this.getSysInfo()
    }
}

getSysInfo() {
    return {
      computerName: os.hostname(),
      computerIp: this.getServerIP(),
      osName: os.platform(),
      osArch: os.arch(),
    };
}

getServerIP() {
    const nets = os.networkInterfaces();
    for (const name of Object.keys(nets)) {
      for (const net of nets[name]) {
        if (net.family === 'IPv4' && !net.internal) {
          return net.address;
        }
      }
    }
}

试一下，和我系统设置里的 ip 一样。这样，服务器的状态数据就都拿到了，可以实现这样的服务监控页面。

总结

RabbitMQ：前端监控系统会收到很多来自用户端的请求，如果直接存入数据库很容易把数据库服务搞挂掉，所以一般会加一个 RabbitMQ 来缓冲。它是生产者往 queue 里放入消息，消费者从里面读消息，之后确认消息收到的流程。当一对多的时候，还要加一个 Exchange 交换机来根据不同的规则转发消息（direct/topic/headers/fanout）。消费者可以设置一个消费的并发上限，保证服务不会因并发过高而崩溃，这就是流量削峰。
请求日志：通过 interceptor 实现了记录请求日志的功能。其中 ip 地址如果被 11-Nginx基础概述转发过，需要取 X-Forwarded-For 的 header 的值，直接用 request-ip 这个包来做。如果想拿到 ip 对应的城市信息，可以用一些免费接口来查询，用 @nestjs/axios 来发送请求（不建议放到请求日志里）。
服务器状态：拿到了服务器的 cpu、内存、磁盘、ip 等信息，通过 node 的 os 模块的 api 以及 node-disk-info 这个包。可以在后台管理系统加一个页面来展示这些信息。

消息队列与监控

为什么前端监控系统要用 RabbitMQ？

RabbitMQ 快速上手

流量削峰的例子

RabbitMQ 整体架构

direct

topic

fanout

headers

如何记录请求日志

根据 ip 拿到城市信息

如何拿到服务器 CPU、内存、磁盘状态

CPU 使用率计算

内存信息

磁盘信息

其他服务器信息

总结

相关文档

消息队列与监控

为什么前端监控系统要用 RabbitMQ？

RabbitMQ 快速上手

流量削峰的例子

RabbitMQ 整体架构

direct

topic

fanout

headers

如何记录请求日志

根据 ip 拿到城市信息

如何拿到服务器 CPU、内存、磁盘状态

CPU 使用率计算

内存信息

磁盘信息

其他服务器信息

总结

相关文档