使用 MQTT 在專案中實現資料收發 | Linux 中國

表中唯一令人不解就是燃料類別中的混合燃料。紐約的大多數天然氣工廠也透過燃燒其它型別的化石燃料發電。在冬季寒潮到來之際，家庭供暖的優先順序高於發電；但這種情況出現的次數不多，（在我們計算中）可以將混合燃料型別看作天然氣型別。

CSV 檔案全天更新。我編寫了一個簡單的資料泵，每隔 1 分鐘檢查是否有資料更新，並將新條目釋出到 MQTT 伺服器的一系列主題上，主題名稱基本與 CSV 檔案有一定的對應關係。資料內容被轉換為 JSON 物件，方便各種程式語言處理。

ny-power/upstream/fuel-mix/Hydro {"units": "MW", "value": 3229, "ts": "05/09/2018 00:05:00"}
ny-power/upstream/fuel-mix/Dual Fuel {"units": "MW", "value": 1400, "ts": "05/09/2018 00:05:00"}
ny-power/upstream/fuel-mix/Natural Gas {"units": "MW", "value": 2144, "ts": "05/09/2018 00:05:00"}
ny-power/upstream/fuel-mix/Other Fossil Fuels {"units": "MW", "value": 4, "ts": "05/09/2018 00:05:00"}
ny-power/upstream/fuel-mix/Wind {"units": "MW", "value": 41, "ts": "05/09/2018 00:05:00"}
ny-power/upstream/fuel-mix/Other Renewables {"units": "MW", "value": 226, "ts": "05/09/2018 00:05:00"}
ny-power/upstream/fuel-mix/Nuclear {"units": "MW", "value": 4114, "ts": "05/09/2018 00:05:00"}

這種直接的轉換是種不錯的嘗試，可將公開資料轉換為公開事件。我們後續會繼續將資料轉換為二氧化碳排放強度，但這些原始資料還可被其它應用使用，用於其它計算用途。

MQTT 主題

主題和主題結構topic structure是 MQTT 的一個主要特色。與其它標準的企業級訊息匯流排不同，MQTT 的主題無需事先註冊。傳送者可以憑空建立主題，唯一的限制是主題的長度，不超過 220 字元。其中 / 字元有特殊含義，用於建立主題的層次結構。我們即將看到，你可以訂閱這些層次中的一些分片。

基於開箱即用的 Mosquitto，任何一個客戶端都可以向任何主題釋出訊息。在原型設計過程中，這種方式十分便利；但一旦部署到生產環境，你需要增加訪問控制串列access control list（ACL）只允許授權的應用釋出訊息。例如，任何人都能以只讀的方式訪問我的應用的主題層級，但只有那些具有特定憑證credentials的客戶端可以釋出內容。

主題中不包含自動樣式automatic schema，也沒有方法查詢客戶端可以釋出的全部主題。因此，對於那些從 MQTT 匯流排消費資料的應用，你需要讓其直接使用已知的主題和訊息格式樣式。

那麼應該如何設計主題呢？最佳實踐包括使用應用相關的根名稱，例如在我的應用中使用 ny-power。接著，為提高訂閱效率，構建足夠深的層次結構。upstream 層次結構包含了直接從資料源獲取的、不經處理的原始資料，而 fuel-mix 層次結構包含特定型別的資料；我們後續還可以增加其它的層次結構。

訂閱主題

在 MQTT 中，訂閱僅僅是簡單的字串匹配。為提高處理效率，只允許如下兩種萬用字元：

為便於理解，下麵給出幾個例子：

ny-power/#  - 匹配 ny-power 應用釋出的全部主題
ny-power/upstream/#  - 匹配全部原始資料的主題
ny-power/upstream/fuel-mix/+  - 匹配全部燃料型別的主題
ny-power/+/+/Hydro - 匹配全部兩次層級之後為 Hydro 型別的主題（即使不位於 upstream 層次結構下）

類似 ny-power/# 的大範圍訂閱適用於低資料量low-volume的應用，應用從網路獲取全部資料並處理。但對高資料量high-volume應用而言則是一個災難，由於絕大多數訊息並不會被使用，大部分的網路頻寬被白白浪費了。

在大資料量情況下，為確保效能，應用需要使用恰當的主題篩選（如 ny-power/+/+/Hydro）儘量準確獲取業務所需的資料。

增加我們自己的資料層次

接下來，應用中的一切都依賴於已有的 MQTT 流並構建新流。第一個額外的資料層用於計算發電對應的二氧化碳排放。

利用美國能源情報署U.S. Energy Information Administration[7] 給出的 2016 年紐約各類燃料發電及排放情況，我們可以給出各類燃料的平均排放率[8]，單位為克/兆瓦時。

上述結果被封裝到一個專用的微服務中。該微服務訂閱 ny-power/upstream/fuel-mix/+，即資料泵中燃料組成情況的原始資料，接著完成計算並將結果（單位為克/千瓦時）釋出到新的主題層次結構上：

ny-power/computed/co2 {"units": "g / kWh", "value": 152.9486, "ts": "05/09/2018 00:05:00"}

接著，另一個服務會訂閱該主題層次結構並將資料打包到 InfluxDB[9] 行程中；同時，釋出 24 小時內的時間序列資料到 ny-power/archive/co2/24h 主題，這樣可以大大簡化當前變化資料的繪製。

這種層次結構的主題模型效果不錯，可以將上述程式之間的邏輯解耦合。在複雜系統中，各個元件可能使用不同的程式語言，但這並不重要，因為交換格式都是 MQTT 訊息，即主題和 JSON 格式的訊息內容。

從終端消費資料

為了更好的瞭解 MQTT 完成了什麼工作，將其系結到一個訊息匯流排並檢視訊息流是個不錯的方法。mosquitto-clients 包中的 mosquitto_sub 可以讓我們輕鬆實現該標的。

安裝程式後，你需要提供伺服器名稱以及你要訂閱的主題。如果有需要，使用引數 -v 可以讓你看到有新訊息釋出的那些主題；否則，你只能看到主題內的訊息資料。

mosquitto_sub -h mqtt.ny-power.org -t ny-power/# -v

只要我編寫或除錯 MQTT 應用，我總會在一個終端中執行 mosquitto_sub。

從網頁直接訪問 MQTT

到目前為止，我們已經有提供公開事件流的應用，可以用微服務或命令列工具訪問該應用。但考慮到網際網路仍佔據主導地位，因此讓使用者可以從瀏覽器直接獲取事件流是很重要。

MQTT 的設計者已經考慮到了這一點。協議標準支援三種不同的傳輸協議：TCP[10]、UDP[11]和 WebSockets[12]。主流瀏覽器都支援 WebSockets，可以維持持久連線，用於實時應用。

Eclipse 專案提供了 MQTT 的一個 JavaScript 實現，叫做 Paho[13]，可包含在你的應用中。工作樣式為與伺服器建立連線、建立一些訂閱，然後根據接收到的訊息進行響應。

// ny-power web console application
var client = new Paho.MQTT.Client(mqttHost, Number("80"), "client-" + Math.random());
// set callback handlers
client.onMessageArrived = onMessageArrived;
// connect the client
client.reconnect = true;
client.connect({onSuccess: onConnect});
// called when the client connects
function onConnect() {
    // Once a connection has been made, make a subscription and send a message.
    console.log("onConnect");
    client.subscribe("ny-power/computed/co2");
    client.subscribe("ny-power/archive/co2/24h");
    client.subscribe("ny-power/upstream/fuel-mix/#");
}
// called when a message arrives
function onMessageArrived(message) {
    console.log("onMessageArrived:"+message.destinationName + message.payloadString);
    if (message.destinationName == "ny-power/computed/co2") {
        var data = JSON.parse(message.payloadString);
        $("#co2-per-kwh").html(Math.round(data.value));
        $("#co2-units").html(data.units);
        $("#co2-updated").html(data.ts);
    }
    if (message.destinationName.startsWith("ny-power/upstream/fuel-mix")) {
        fuel_mix_graph(message);
    }
    if (message.destinationName == "ny-power/archive/co2/24h") {
        var data = JSON.parse(message.payloadString);
        var plot = [
            {
                x: data.ts,
                y: data.values,
                type: 'scatter'
            }
        ];
        var layout = {
            yaxis: {
                title: "g CO2 / kWh",
            }
        };
        Plotly.newPlot('co2_graph', plot, layout);
    }