手把手用 UDP 实现 Node 服务日志归集（附完整源码）

日志归集：顾名思义，就是把日志归集起来。

在开发 Node 服务的时候，咱们常常会打印各类日志，好比 info, error 日志。

若是咱们已经将服务已经部署了不少机器上，这个时候若是须要查询昵称为”张三“的日志流水，那就很痛苦了，机器越多越难搞，由于不知道昵称为”张三“的日志流水到底在哪一台，甚至可能还分散在多台服务器。

所以，咱们须要作日志归集，这样咱们在一台机器上就能够查看全部的日志了。

本文目录：

• 使用 dgram 实现日志归集
• UDP 数据包 和 Unix 数据报
• 再看 dgram

第一步：收集日志

服务是部署在多台服务器上，每台机器上都会打印日志，最终咱们要归集起来到一台机器上展现，那么首先须要把每台服务器上的日志拿到，通常有两种实现方式。

使用日志 logger 组件

若是整个系统日志输出，咱们是使用统一的自定义 logger 组件，好比执行 logger.info(), logger.error() 来输出日志的话。那么咱们就很容易收集日志了，直接在 logger 的 info , error, warn 等方法里面收集便可。

好比以下，在 logger 日志对象里面，增长存储到 logList 临时对象里面。

const logList=[];
['info','warn','error'].forEach((type)=>{  this[type] = ()=>{  // do something  logList.push({type:type,data:Array.from(arguments).join(" ")});  } }); 复制代码

这里咱们须要用 logList 数组保存起来，是为了数据能批量发送，在指定的间隔时间（好比 1s）,发送一第二天志，而不是每执行一次 log，就须要发送一次请求。

扫描文件上报

有些使用了第三方框架，写日志，并非通过咱们自定义的 logger 组件，这个时候，咱们能够采起读取文件的方式来收集日志数据。实现以下：

const fsExtra = require('fs-extra');
// 监听文件 path 的变化 fsExtra.watchFile(path, (curr, prev) => {  // 有时候响应不稳定，会触发两次，第二次触发的时候，对象里面的时间戳会相等，因此判断当前时间戳小于等于上一次时间戳可过滤多余的监听。  if (curr.mtime <= prev.mtime) return;  // 为了提升性能，咱们采起只读的方式来 open  fsExtra.open(path, 'r', function(error, fd) {  if (error) {  logger.error('open log error', error);  return;  }  // 建立一个缓冲，长度为（当前文件大小-文件上一个状态的大小） 因为这里咱们已经明确获取到长度了，因此建立buffer可使用性能更好的 allocUnsafe  buffer = Buffer.allocUnsafe(curr.size - prev.size);  fsExtra.read(fd, buffer, 0, (curr.size - prev.size), prev.size, function(err, bytesRead, buffer) {  if (err) {  logger.error('read log error', error);  return;  }  const data = buffer.toString();  logList.push({data:data});  });  }); 复制代码

逻辑也比较简单，监听日志文件（若是须要监听文件夹，在可使用 watch）。当监听到文件变动的时候，已只读的方式打开文件，而后使用 read 函数读取文件的从起始到结束为止的内容。

第二步：客户端发送 UDP 数据包

UDP 数据包的特色是无链接，结构简单，对系统资源消耗少。缺点是数据不保证正确，且数据包到达前后顺序不保证。

对于咱们的日志归集，显然不须要保证正确，且前后顺序影响不大，咱们在每一条数据消息里面携带发送消息客户端的时间戳便可。因此咱们选择 UDP 数据包。

这里到底使用 udp4 仍是 udp6 ,取决于咱们服务器，对应咱们熟悉的是 IPV4 和 IPV6。这里使用 udp4。

const dgram = require('dgram');
const clientSocket = dgram.createSocket('udp4'); // 原创 udp 日志服务器的 ip 和 port，可能会是配置下发的方式来肯定。若是是明确的一台服务器，也能够写死。 let remoteUDPServerIP = 'xxx,xxx,xxx,xxx'; let remoteUDPServerPort = 'xxx'; // 当前 udp 服务的 ip clientSocket.bind(7777); setInterval(()=>{  if(logList.length==0) return;  const data = logList.join("\r\n");  logList = [];  clientSocket.send(data, 0, data.length, remoteUDPServerPort, remoteUDPServerIP); },1000);  复制代码

开启定时器，扫描存储的日志列表。一般来讲，若是咱们的日志数据 logList ，是由咱们本身定义的 logger 获取的，那么咱们须要先用数组存起来，而后定时发送 UDP 数据包到 server 服务器上，达到批量发送的目的，减小发送频率。

可是若是咱们是使用读取文件的方式来拿到须要上报的日志的话，咱们能够监听到文件变化而后直接上报便可，无需使用定时器发送，由于会在写日志文件的时候，已经作了批量写入了（通常会是 1s 更新一次）。

第三步：服务器接受 UDP 数据包

使用 dgram 建立服务端，而后接受客户端发送的消息，最终写到统一的日志文件里面。

const path = require("path");
const fsExtra = require('fs-extra'); const dgram = require('dgram'); const serverSocket = dgram.createSocket('udp4');  serverSocket.on('message', function(msg, rinfo){  handleMsg(msg,rinfo.address); }); // 服务端端口 serverSocket.bind(7777);  var fp = "./logs"+path.sep+path.sep+msg.type; fsExtra.ensureDir(fp);  function handleMsg(msg,address){  try{  msg = typeof msg=="object"?JSON.parse(msg):{msg:msg};  }catch(e){  console.log(e);  }  if(!msg.type) msg.type="info";   // 组装上报参数  let filepath = fp+path.sep+ getDate()+".log";  let content = [];  content.push(getDate(true));  content.push(address);  content.push(typeof msg.msg=="object"?JSON.stringify(msg.msg):msg.msg);  content.push('\r\n');  // 写入日志文件  fs.appendFile(filepath, content.join(" "), err=> {  if(err){  fs.appendFile(".logs/white-error.log", filepath+"\t"+msg, e=>{});  }else{  // report   }  }); } function getDate(time) {  let date = new Date();  if(time){  let hour = date.getHours();  let minute = date.getMinutes();  let second = date.getSeconds();  let miseconds = date.getMilliseconds();  return [hour, minute, second,miseconds].map(formatNumber).join(':');  }  let year = date.getFullYear();  let month = date.getMonth() + 1;  let day = date.getDate();  let p = [year, month, day].map(formatNumber).join('-');  return p; } function formatNumber(n) {  n = n.toString();  return n[1] ? n : '0' + n; } 复制代码

这里大伙看下应该能看懂了。建立 7777 端口的服务。而后监听到消息，则当即写入文件。

第四步：查询日志

至此，咱们就实现了一个简单的 UDP 数据包的日志归集功能。无论部署多少台服务器。咱们只须要在 ./logs/ 目录下就能够查看日志了。执行 tail 命令就能够查看各个服务器归集过来的日志了，固然时间可能会有错乱，可是以每条消息的时间为准便可。

$ tail -f logs/info/2020-04-19.log
复制代码

UDP 数据包 和 Unix domain socket 数据报

UDP，全称 User Datagram Protocol, 即用户数据报协议。UDP 和 TCP 二者的具体差别就不罗列了，有兴趣能够去了解下。主要差别以下：

这里咱们主要看和 unix 协议的差别。

Unix domain socket（UDS）：是 unix 服务器进程之间本地通讯 IPC 的一种。提供了两套协议：字节流套接口（相似 TCP ）和数据报套接口（相似 UDP )。

unix 数据报套接口与 UDP 的区别主要在于 UDS 只能作本机 IPC， 而 UDP 能够是本机也能够是远程机器通讯。

若是仅仅是本机通讯的话，推荐直接使用 UDS。主要优点有如下两个：

• UDS 协议的数据报不会出现丢失乱序的状况。
• UDS 协议性能更优，不会跑到链路层，不会作数据报校验。

固然若是为了可扩展，后续迁移到不一样的机器，也可使用 UDP 来实现。

UDS 协议可使用 Node 的 unix-dgram 组件来实现。

因为 UDS 是本机通讯，因此不须要端口和 IP, 直接使用文件系统表示的路径名来标识。好比 /path/to/socket 这个系统路径。若是须要建立多个 server, 则指定不一样的路径便可。咱们来看下具体的使用代码。

服务接收端：

const unix = require('unix-dgram');
const server = unix.createSocket('unix_dgram', function(buf) {  console.log('received ' + buf); }); server.bind('/path/to/socket'); 复制代码

客户发送端：

const message = Buffer('ping');
const client = unix.createSocket('unix_dgram'); client.send(message, 0, message.length, '/path/to/socket'); 复制代码

也能够自行定义其余 EventsListener,

const unix = require('unix-dgram');
const client = unix.createSocket('unix_dgram'); client.on('error', function(err) {  console.error(err); }); client.on('connect', function() {  console.log('connected');  client.on('congestion', function() {  /* The server is not accepting data */  });  client.on('writable', function() {  /* The server can accept data */  });  var message = new Buffer('ping');  client.send(message); }); client.connect('/tmp/server'); 复制代码

使用方式仍是比较简单的，相信大伙看看就能懂了。

再看 dgram

dgram 是 Nodejs 提供的用于发送 UDP 数据报的模块。

dgram 提供了很是丰富的 api，能够参考文档：nodejs.cn/api/dgram.h…;

主要功能是能够实现 ”单播”、“广播”和“组播”消息。

单播

单播即单向通讯，客户端向服务端发送消息，本文中的日志归集就是单播实现。参考上面实现。

广播

广播顾名思义，就是广撒网，非点对点通讯。把数据发送给本地子网上的全部的机器，即广播。要实现广播，首先要获取到广播地址。好比我在终端输入：ifconfig，以下显示的 broadcast 就是广播地址。

知道了广播地址，则能够在服务端向子网内全部机器发送广播消息了。

const dgram = require("dgram");
const serverUDP = dgram.createSocket("udp4"); serverUDP.on("listening", () => {  console.log("socket正在监听...");  server.setBroadcast(true);  server.setTTL(64);  setTimout(() => {  server.send("你们好啊，我是服务端.", 7778, "192.168.0.255")  }, 1000) }) serverUDP.on("message", (msg, rinfo) => {  console.log(`msg from client ${rinfo.address}:${rinfo.port}`); }) server.bind(7777); 复制代码

这里我建立了一个服务端，监听 7777 端口。向广播地址：192.168.0.255 端口为 7778 的全部客户端发送广播消息，因而属于同一个广播区域段的客户端的 7778 端口都会收到服务端的消息。

默认是单播，须要广播消息，设置 setBroadcast 为 true 便可。

IP_TTL : 表示存活时间，每通过个路由器或者网关都会减小 TTL 数值，若是 TTL 被一个路由器减小到 0，这个数据报将不会继续转发。

组播

组播报文的目的地址使用D类IP地址， D类地址不能出如今IP报文的源IP地址字段。

224.0.0.0～224.0.0.255为预留的组播地址（永久组地址），地址224.0.0.0保留不作分配，其它地址供路由协议使用； 224.0.1.0～224.0.1.255是公用组播地址，能够用于Internet； 224.0.2.0～238.255.255.255为用户可用的组播地址（临时组地址），全网范围内有效； 239.0.0.0～239.255.255.255为本地管理组播地址，仅在特定的本地范围内有效。

下面已 224.1.1.1 组播地址测试。其余代码通广播。

服务端：

const addr = '224.1.1.1';
server.on("listening",()=>{  console.log("socket正在监听中.....");  server.addMembership(addr);  server.setMulticastTTL(64);  setTimout(()=>{  server.send('你们好啊，我是服务端.',7778,addr);  },1000) }) 复制代码

客户端：

const dgram = require("dgram");
const client = dgram.createSocket("udp4"); const addr = '224.1.1.1'; client.on("listening", () => {  console.log("socket正在监听...");  client.addMembership(addr); }) client.on("message", (msg, rinfo) => {  console.log(`msg from server:${msg},addr:${rinfo.address}`); }) client.bind(7778) 复制代码

到此为止，简易的日志服务系统已经开放完成，在实际中，咱们可能不会直接这样使用，通常有两种方式：

• 使用自研的 node 框架或者统一的日志组件，而后由框架容器或者组件去实现全部的日志收集，上报和展现。
• 使用统一的日志服务，好比 c++ 开发的日志服务，按照配置和设定的规则，统一收集日志信息，上报和展现。

对于业务开发者来讲，可能就只须要在项目里面配置下，就能享受流畅的日志查看了。

然而咱们只有在工做中跳出仅仅做为使用者的角度，去探索各个系统的实现原理，这样才能游刃有余。

源码 https://github.com/antiter/udp-log

欢迎关注个人微信公众号，一块儿作靠谱前端！