搭建前端監控系統（五）Nodejs怎麼搭建消息隊列

　　怎樣定位前端線上問題，一直以來，都是很頭疼的問題，由於它發生於用戶的一系列操做以後。錯誤的緣由可能源於機型，網絡環境，接口請求，複雜的操做行爲等等，在咱們想要去解決的時候很難復現出來，天然也就沒法解決。 固然，這些問題並不是不能克服，讓咱們來一塊兒看看如何去監控並定位線上的問題吧。 

 

　　這是搭建前端監控系統的第五章，主要是介紹如何處理日誌高併發上傳的狀況，跟着我一步步作，你也能搭建出一個屬於本身的前端監控系統。

　　若是感受有幫助，或者有興趣，請關注 or Star Me 。 

 

　　具體效果： 前端監控系統

 

　　隨着監控日誌蒐集的內容愈來愈多，終於有一天，因爲公司的一波推文，致使了日誌的瞬間流量達到歷史新高，以致於mysql的鏈接數過多，系統崩潰。 固然，做爲日誌上傳的服務器，這個是必然會發生的狀況，只是遲早的問題。 既然出現了併發問題，那麼咱們就着手來處理吧。日誌上傳如何緩解高併發的狀況呢？咱們分爲三個小點來處理。

• 　　增長日誌上傳的時間間隔

　　正如咱們所知，日誌上傳的時間間隔越長，用戶在這個間隔內離開的概率就會越大，日誌的漏傳量就會增長，而後會致使日誌的準確度下降。由於咱們的探針是安插在瀏覽器內的，用戶隨時都有可能關掉，因此，理論上講間隔越短越好，但這並不現實。因此這個須要在服務器的承受能力和日誌的準確率之間作個權衡。由具體狀況而定

• 　　移除探針代碼裏冗餘的參數，縮短參數名字的長度 

　　另一點，每臺服務器的硬盤有限，帶寬有限，若是參數名字太長，參數內容冗餘，對服務器的硬盤和帶寬都是一種極大的浪費。雖然每條日誌都不起眼，可是日誌起量了之後，就是會是一筆很是龐大的開銷。

• 　　Nodejs + RabbitMq 搭建消息隊列，緩解瞬間併發量　　

　　對於一個前端來講，要把消息隊列搭建起來還確實費了一番周折。 

　　1）  ubantu16 安裝RabbitMQ服務軟件包，不少教程都要求安裝erlang, 可是更新apt之後，直接執行安裝命令，會自動安裝erlang的核心組件的。（erlang始終沒法成功安裝，真心累。）

$ apt-get update
$ apt-get install rabbitmq-server // 安裝
$ rabbitmq-plugins enable rabbitmq_management // 啓動插件，瀏覽器才能訪問

　　正常狀況下是直接成功的，直接訪問ip端口號就能夠打開了 http://IP:15672, 以下圖：

　　2）如今咱們須要一個有效的登陸名和密碼，執行以下命令

$ rabbitmqctl add_user username password  // 設置用戶名密碼
$ rabbitmqctl set_user_tags username administrator  // 設置爲管理員身份
$ rabbitmqctl set_permissions -p / username ".*" ".*" ".*"  //爲用戶設置讀寫等權限 

　　OK, 如今咱們登陸進來就是這樣的界面，如此消息隊列服務咱們算是搭建完成了。

　　3）消息服務啓動了，那麼如何存消息，如何取消息呢？以下圖所示：

　　

　　我可以接觸到的關於消息隊列的應用場景實在有限，因此不能介紹更復雜的內容，大體的思惟邏輯如上圖1：有消息進來，先存入消息隊列裏，另外一端再從隊列去取出來，完成接下來的工做。從代碼的角度來看如上圖2：就是一個生產者和消費者的模式，生產者不停的向消息隊列裏生產消息，消費者在有須要的時候，從消息隊列裏取消息， 一旦完成消費，隊列裏便移除這個消息。消息的生產者和消費者互相沒有感知，生產者產生過剩的消息都存放在消息隊列裏，由消費者慢慢消耗。以此來削峯填谷，達處處理高併發的目的。固然這都是個人淺顯理解，可是也足以知足目前日誌上傳的需求了。

　　OK、理論說完了，具體如何實現呢？

　　

let amqp = require('amqplib');

module.exports = class RabbitMQ {
  constructor() {
    this.hosts = ["amqp://localhost"]; 
    this.index = 0;
    this.length = this.hosts.length;
    this.open = amqp.connect(this.hosts[this.index]);
  }
　// 消息生產者
  sendQueueMsg(queueName, msg, errCallBack) {
    let self = this;
    self.open
      .then(function (conn) {
        return conn.createChannel();
      })
      .then(function (channel) {
        return channel.assertQueue(queueName).then(function (ok) {
          return channel.sendToQueue(queueName, new Buffer.from(msg), {
            persistent: true
          });
        })
          .then(function (data) {
            if (data) {
              errCallBack && errCallBack("success");
              channel.close();
            }
          })
          .catch(function () {
            setTimeout(() => {
              if (channel) {
                channel.close();
              }
            }, 500)
          });
      })
      .catch(function () {
　　　　　　// 這裏嘗試備用鏈接，我就一個，因此就處理了
      });
  }
　
  // 消息消費者
  receiveQueueMsg(queueName, receiveCallBack, errCallBack) {
    let self = this;

    self.open.then(function (conn) {
        return conn.createChannel();
      }).then(function (channel) {
        return channel.assertQueue(queueName).then(function (ok) {
            return channel.consume(queueName, function (msg) {
              if (msg !== null) {
                let data = msg.content.toString();
                channel.ack(msg);
                receiveCallBack && receiveCallBack(data);
              }
            }).finally(function () { });
          })
      })
      .catch(function (e) {
        errCallBack(e)
      });
  }
}

 

消息隊列測試：每隔5秒發送一條消息，每隔5秒取出一條消息，成功

var mq = new RabbitMQ()
setInterval(function () {
  mq.sendQueueMsg("queue1", "這是一個隊列消息", function (err) {
    console.log(err)
  })
}, 5000)

setInterval(function () {
  mq.receiveQueueMsg("queue1", function (msg) {
    console.log(msg)
  }, function (error) {
    console.log(error)
  })
}, 5000)

 

RabbitMq消息隊列使用中遇到的坑：

①   var mq = new RabbitMQ() 屢次建立RabbitMQ對象，致使connections, channels, memory 暴增，服務器很快掛掉

②  生產者的channel忘記close, 致使channel太多，服務器超負荷

③   消費者的channel被close掉了，永遠只能接收到一條消息，消息隊列很快爆掉

最後是消息隊列運行的狀態：

 

OK、通過了這麼一番處理，咱們的日誌上傳應該可以承受住必定量的併發了，讓咱們拭目以待吧。

 

上一章：搭建前端監控系統（四）接口請求異常監控篇

 

參考： 架構設計之NodeJS操做消息隊列RabbitMQ  Ubuntu上安裝和使用RabbitMQ