使用 AWS SQS 在 Node.js 中進行消息隊列
簡介
隨著現代軟件系統複雜性的增加，需要分解已經超出其初始規模的系統。系統複雜性的增加使得維護、更新和升級變得更加困難。
這為微服務鋪平了道路，允許將龐大的單體系統分解為更小的服務，這些服務鬆散耦合但交互以提供初始單體解決方案的全部功能。松耦合提供了敏捷性並簡化了維護和添加新功能的過程，而無需修改整個系統。
隊列系統正是在這些微服務架構中 可以方便地促進構成整個設置的各個服務之間的通信。
在這篇文章中，我們將深入研究隊列系統，尤其是 Amazon 的 Simple Queue Service，並演示我們如何在微服務環境中利用其功能。
什麼是消息隊列？
在互聯網和電子郵件出現之前，人們遠距離交流主要是通過書信交流。這些信件包含要分享的信息，並被郵寄到當地郵局，然後從那裡將它們轉移到收件人的地址。
這可能因地區而異，但想法是相同的。人們在繼續生活的過程中委託中介為他們傳遞信息。
當一個系統被分解成更小的組件或服務，這些組件或服務需要協同工作時，它們需要根據各個服務的功能進行通信並將信息從一個服務傳遞到另一個服務。
消息隊列通過充當微服務的“郵局服務”來促進這一過程。消息被放入隊列中，目標服務接收並處理髮送給它們的消息。消息可以包含任何內容 - 例如有關要採取的步驟、要操作或保存的數據或要執行的異步作業的說明。
消息隊列是一種允許系統組件以異步方式通信和交換信息的機制。這意味著鬆散耦合的系統不必等待對它們發送的消息的即時反饋，它們可以騰出來繼續處理其他請求。當時機成熟，需要響應時，服務可以在消息隊列中尋找響應。
以下是一些流行的消息隊列或代理的示例：
- 亞馬遜簡單隊列服務 - 這是本文的重點
- RabbitMQ - 開源並提供異步消息傳遞功能
- 阿帕奇卡夫卡 - 這是一個支持發布/訂閱交互模式的分佈式流媒體平台
- 其他包括 Apache RocketMQ , NSQ , 和 HornetQ
消息隊列用例
並非每個系統都需要消息隊列，但在某些情況下，它們值得付出努力和資源來設置和維護它們。如果使用得當，消息隊列在幾個方面都有優勢。
首先，消息隊列通過在松耦合系統中提供通信機制來支持大系統的解耦。
通過在服務失敗時維護狀態來使用消息隊列來支持冗餘。當一個失敗或有故障的服務恢復操作時，它本來要處理的所有操作仍將在隊列中，它可以拾取它們並繼續處理原本可能丟失的事務。
消息隊列有助於批量操作，例如發送電子郵件或將記錄插入數據庫。批量指令可以保存在一個隊列中，所有指令同時按順序處理，而不是一個一個地處理，效率可能很低。
排隊系統還有助於確保操作的一致性，方法是確保它們按照接收到的順序執行。當系統的特定組件或服務已被複製以處理增加的負載時，這一點尤其重要。這樣，系統將很好地擴展以處理負載，並確保處理的事務是一致且有序的，因為所有復制的服務都將從充當單一事實來源的消息隊列中獲取它們的指令。
亞馬遜簡單隊列服務 - SQS
與 Amazon Web Services 的大多數其他產品一樣，Simple Queue Service (SQS) 是一種消息隊列解決方案，由 Amazon 進行分佈式和完全管理，就像通過 Chalice 進行的無服務器計算一樣。
SQS 允許我們在軟件組件之間發送和接收消息或指令，使我們能夠在我們的系統中實現和擴展微服務，而無需設置和維護排隊系統。
與其他 AWS 服務一樣，SQS 可根據需求動態擴展，同時確保通過（可選）消息加密傳遞的數據的安全性。
演示項目
探索 Amazon Simple Queue Service ，我們將在 Node.js 中創建一個解耦系統，每個組件將通過從 SQS 發送和檢索消息來與其他組件交互。
由於我們是一個小型組織，沒有足夠的帶寬來處理收到的訂單，因此我們將有一項服務來接收用戶的訂單，而另一項服務將在當天的某個時間將當天發布的所有訂單發送到我們的電子郵件收件箱。批處理的一天。所有訂單都將存儲在隊列中，直到我們的第二個服務收集並發送到我們的電子郵件收件箱。
我們的微服務將由簡單的 Node.js API 組成，一個用於接收用戶的訂單信息，另一個用於向用戶發送確認電子郵件。
通過消息隊列異步發送電子郵件確認將允許我們的訂單服務在負載的情況下繼續接收訂單，因為它不必擔心發送電子郵件。
另外，萬一郵件服務宕機，一旦恢復，它會繼續從隊列中發送郵件，因此我們不必擔心丟失訂單。
亞馬遜網絡服務
對於這個項目，我們需要一個活躍且有效的 AWS 賬戶，我們可以在 AWS 主頁上註冊。 AWS 要求我們不僅提供一些個人詳細信息，還要求我們提供賬單詳細信息。為了避免為此演示項目收費，我們將使用 AWS 免費套餐進行測試和開發。
我們還需要安裝 AWS CLI 工具，以便與我們機器上的 AWS 資源進行交互。可以在此處找到在多個平台上安裝 AWS CLI 工具的說明。
使用 AWS CLI 工具後，我們可以前往 AWS 控制台 在我們的個人資料下拉列表中有一個名為“我的安全憑據的部分 "。在這裡，我們將能夠創建將在與 AWS 控制台交互時使用的憑據。
Amazon CLI 工具也將使用這些憑證，我們將通過運行配置該工具：
$ aws configure
我們將收到提示填寫我們的 Access Key ID , Secret Access Key ，以及默認區域和輸出格式。最後兩個是可選的，但我們需要從 AWS 控制台儀表板獲得的訪問密鑰和秘密。
在我們的 AWS 賬戶啟動並運行並配置 AWS CLI 後，我們可以通過導航到 SQS 主頁來設置我們的 AWS 簡單隊列服務。
正如我們在下面的屏幕截圖中看到的，在將我們的隊列名稱指定為 nodeshop.fifo 之後 我們會看到兩個隊列選項：
我們可以選擇標準隊列 或 FIFO 隊列 .標準隊列不維護它接收到的消息的順序，更適合將吞吐量優先於事件順序的項目。
另一方面，FIFO 隊列維護接收到的消息的順序，並且它們也以相同的先進先出進行檢索 順序。
鑑於我們將構建一個迷你購物平台，因此保持請求的順序很重要，因為我們希望按照購買順序向人們出售商品。一旦我們選擇了我們需要的隊列類型，我們就可以修改隊列的一些額外配置：
我們可以配置消息從隊列中自動刪除之前的時間和消息的大小，以及其他選項。現在，我們將使用默認值配置我們的 FIFO 隊列。隊列準備就緒後，我們現在可以創建 Node.js API 來讀取和寫入我們的 Amazon SQS FIFO 隊列。
設置 Node.js 和 NPM
在 Node.js 官方網站上可以找到在多個平台上設置 Node.js 的說明。 Node 包管理器 (NPM) 隨 Node.js 一起提供，我們可以按如下方式驗證我們的安裝：
# Node.js version
$ node -v
v12.12.0
# NPM version
$ npm -v
6.11.3
下一步是按如下方式設置我們的文件夾：
# create folder and move into it
$ mkdir nodeshop_apis && cd $_
# create the orders and emails services folders
$ mkdir orderssvc emailssvc
設置節點 API
我們將構建 orders 服務第一，因為它是接收用戶訂單並將信息發佈到我們隊列中的服務。我們的 emails 然後服務將從隊列中讀取並發送電子郵件。
我們將初始化一個 Node.js 項目並安裝 Express.js 框架，我們將使用它來構建我們的極簡 API。我們還將安裝 body-parser 中間件來為我們處理請求數據並對其進行驗證。
在我們的根文件夾中實現這一點：
# initialize node project
$ npm init
# install express and body-parser
$ npm install express body-parser --save
免費電子書：Git Essentials
查看我們的 Git 學習實踐指南，其中包含最佳實踐、行業認可的標準以及隨附的備忘單。停止谷歌搜索 Git 命令並真正學習 它！
一次 Express 和 body-parser 已安裝，它們將自動添加到我們的 package.json 的依賴項部分 文件感謝 --save 選項。
由於我們將同時運行多個服務，因此我們還將安裝 npm-run-all 包來幫助我們同時啟動所有服務，而不必在多個終端窗口中運行命令：
$ npm install npm-run-all --save
使用 npm-run-all 安裝好了，現在讓我們調整一下 scripts 在我們的 package.json 中輸入 文件包含啟動我們服務的命令和一個運行它們的命令：
{
// Truncated for brevity...
"scripts": {
"start-orders-svc": "node ./orderssvc/index.js 8081",
"start-emails-svc": "node ./emailssvc/index.js",
"start": "npm-run-all -p -r start-orders-svc"
},
// ...
}
我們將添加命令 start-orders-svc 和 start-emails-svc 運行我們的 orders 和 emails 分別服務。然後我們將配置 start 命令使用 npm-run-all 執行它們 .
有了這個設置，運行我們所有的服務就像執行以下命令一樣簡單：
$ npm start
我們可以創建我們的 orders index.js 中的 API 文件如下：
const express = require('express');
const bodyParser = require('body-parser');
const port = process.argv.slice(2)[0];
const app = express();
app.use(bodyParser.json());
app.get('/index', (req, res) => {
res.send("Welcome to NodeShop Orders.")
});
console.log(`Orders service listening on port ${port}`);
app.listen(port);
將所需的庫添加到我們的 Express app 後 ，“/index”端點將通過簡單地發送歡迎消息來響應。最後，API 將監聽我們在啟動時指定的端口。
我們將通過運行 npm start 來啟動應用程序 使用 Postman 應用程序命令並與我們的 API 交互：
我們將實現 emails 稍後服務。現在，我們的 orders 服務設置好了，我們現在可以實現我們的業務邏輯了。
實現：訂單服務
為了實現我們的業務邏輯，我們將從 orders 開始 該服務將接收我們的訂單並將其寫入我們的 Amazon SQS 隊列。
我們將通過引入一個新的路由和控制器來處理來自最終用戶的訂單輸入，並將訂單數據發送到我們的 Amazon SQS 隊列來實現這一點。
在實現控制器之前，我們需要安裝 Amazon SDK for Node.js：
$ npm install aws-sdk --save
我們的新“/order”端點將接收包含訂單數據的有效負載，並使用 AWS 開發工具包將其發送到我們的 SQS 隊列：
// ./orderssvc/index.js
//
// Code removed for brevity...
//
// Import the AWS SDK
const AWS = require('aws-sdk');
// Configure the region
AWS.config.update({region: 'us-east-1'});
// Create an SQS service object
const sqs = new AWS.SQS({apiVersion: '2012-11-05'});
const queueUrl = "SQS_QUEUE_URL";
// the new endpoint
app.post('/order', (req, res) => {
let orderData = {
'userEmail': req.body['userEmail'],
'itemName': req.body['itemName'],
'itemPrice': req.body['itemPrice'],
'itemsQuantity': req.body['itemsQuantity']
}
let sqsOrderData = {
MessageAttributes: {
"userEmail": {
DataType: "String",
StringValue: orderData.userEmail
},
"itemName": {
DataType: "String",
StringValue: orderData.itemName
},
"itemPrice": {
DataType: "Number",
StringValue: orderData.itemPrice
},
"itemsQuantity": {
DataType: "Number",
StringValue: orderData.itemsQuantity
}
},
MessageBody: JSON.stringify(orderData),
MessageDeduplicationId: req.body['userEmail'],
MessageGroupId: "UserOrders",
QueueUrl: queueUrl
};
// Send the order data to the SQS queue
let sendSqsMessage = sqs.sendMessage(sqsOrderData).promise();
sendSqsMessage.then((data) => {
console.log(`OrdersSvc | SUCCESS: ${data.MessageId}`);
res.send("Thank you for your order. Check you inbox for the confirmation email.");
}).catch((err) => {
console.log(`OrdersSvc | ERROR: ${err}`);
// Send email to emails API
res.send("We ran into an error. Please try again.");
});
});
AWS SDK 要求我們構建一個有效負載對象，指定我們發送到隊列的數據，在我們的例子中，我們將其定義為 sqsOrderData .
然後我們將此對像傳遞給 sendMessage() 該函數將使用我們用於配置 AWS CLI 的憑證將我們的消息發送到隊列。最後，我們等待響應並通知用戶他們的訂單已成功收到，他們應該檢查電子郵件確認。
測試 orders 服務，我們運行命令 npm start 並將以下有效負載發送到 localhost:8081/order ：
{
"itemName": "Phone case",
"itemPrice": "10",
"userEmail": "[email protected]",
"itemsQuantity": "2"
}
這會將我們的訂單提交到 orders 服務，消息將從那裡發送到我們的 SQS 隊列。我們可以通過AWS控制台查看SQS隊列中的訂單，如圖：
我們的 orders 服務已經能夠接收用戶的訂單並成功將數據發送到我們在簡單隊列服務上的隊列 .
實現：電子郵件服務
我們的 orders 服務已準備就緒，並且已經收到用戶的訂單。 emails 服務將負責讀取存儲在隊列中的消息並向用戶發送確認電子郵件。下訂單時不會通知此服務，因此必須不斷檢查隊列是否有任何新訂單。
確保我們的 emails 服務不斷檢查新訂單，我們將使用 sqs-consumer 該庫將持續並定期檢查新訂單並將電子郵件發送給用戶。 sqs-consumer 一旦從隊列中成功讀取消息，也會從隊列中刪除消息。
我們將從安裝 sqs-consumer 開始 庫通過運行以下命令：
$ npm install sqs-consumer --save
現在我們可以實現 emails 服務如下：
const AWS = require('aws-sdk');
const { Consumer } = require('sqs-consumer');
// Configure the region
AWS.config.update({region: 'us-east-1'});
const queueUrl = "SQS_QUEUE_URL";
// Configure Nodemailer to user Gmail
let transport = nodemailer.createTransport({
host: 'smtp.googlemail.com',
port: 587,
auth: {
user: 'Email address',
pass: 'Password'
}
});
function sendMail(message) {
let sqsMessage = JSON.parse(message.Body);
const emailMessage = {
from: 'sender_email_adress', // Sender address
to: sqsMessage.userEmail, // Recipient address
subject: 'Order Received | NodeShop', // Subject line
html: `<p>Hi ${sqsMessage.userEmail}.</p. <p>Your order of ${sqsMessage.itemsQuantity} ${sqsMessage.itemName} has been received and is being processed.</p> <p> Thank you for shopping with us! </p>` // Plain text body
};
transport.sendMail(emailMessage, (err, info) => {
if (err) {
console.log(`EmailsSvc | ERROR: ${err}`)
} else {
console.log(`EmailsSvc | INFO: ${info}`);
}
});
}
// Create our consumer
const app = Consumer.create({
queueUrl: queueUrl,
handleMessage: async (message) => {
sendMail(message);
},
sqs: new AWS.SQS()
});
app.on('error', (err) => {
console.error(err.message);
});
app.on('processing_error', (err) => {
console.error(err.message);
});
console.log('Emails service is running');
app.start();
我們將創建一個新的 sqs-consumer 使用 Consumer.create() 應用程序 函數並提供查詢 URL 和處理從 SQS 隊列獲取的消息的函數。
在我們的例子中，函數 sendMail() 將從隊列中獲取消息，提取用戶訂單的詳細信息，然後使用 Nodemailer 向用戶發送電子郵件 .如果您想了解更多信息，請查看我們在 Node.js 中發送電子郵件的文章。
我們的 emails 服務現已準備就緒。要將其集成到我們的執行腳本中，我們只需修改 scripts 我們的 package.json 中的選項 ：
{
// Truncated for brevity...
"scripts": {
"start-orders-svc": "node ./orderssvc/index.js 8081",
"start-emails-svc": "node ./emailssvc/index.js",
// Update this line
"start": "npm-run-all -p -r start-orders-svc start-emails-svc"
},
// ...
}
當我們通過 orders 提交新訂單時 服務，我們在收件箱中收到以下電子郵件：
結論
在這篇文章中，我們使用 Node.js 和 Express 創建了一個 API，用於接收用戶的訂單並將訂單詳細信息發佈到我們在 AWS 上的 SQS 隊列。然後，我們構建了另一個服務來獲取發佈在隊列中的消息，並向發布訂單的用戶發送確認電子郵件。
我們將排序邏輯與電子郵件管理邏輯分開，並使用消息隊列系統將這兩個服務組合在一起。這樣我們的 orders emails 服務可以處理訂單放置 服務將電子郵件發送給用戶。
該項目的源代碼可在 GitHub 上找到。/
