เชี่ยวชาญ Message Queues ด้วย Node.js & Redis: เพิ่มประสิทธิภาพ Web App

เมื่อสร้าง large scale web applications ความเร็วเป็น priority หลัก ผู้ใช้ไม่อยากรอ response นาน และไม่ควรรอ แตกระบวนการบางอย่างใช้เวลา และไม่สามารถเร็วขึ้นหรือ eliminate ได้

Message queues แก้ปัญหานี้โดยให้ additional branch เพิ่มใน request-response journey ปกติ Branch เพิ่มเติมนี้ช่วยให้ users ได้ immediate responses และ time-consuming processes ทำงานด้านข้าง ทุกคนกลับบ้านด้วยความสุข

บทความนี้โฟกัสอธิบาย message queues คืออะไร และเริ่มต้นใช้งานด้วยการสร้าง application เรียบง่าย คุณควรคุ้นเคย Node.js basics และติดตั้ง Redis ไว้ local หรือ cloud instance เรียนรู้ติดตั้ง Redis ที่นี่

Queue คืออะไร?

Queue เป็น data structure ให้เก็บ entities ในลำดับ Queues ใช้หลักการ first-in-first-out (FIFO)

Concept ของ queues ใน computer science เหมือน queues ในชีวิตประจำวัน ที่คนจับคิวเพื่อรับสิ่งต่างๆ คุณเข้าคิวจากหลัง รอจนถึงคุณ แล้วออกจากคิวหน้าหลังได้รับบริการ

ใน computer science เมื่อ process เช่น API request กำลังรัน และคุณต้องนำ task 某 (เช่นส่งอีเมล) ออกจาก current flow คุณ push มันเข้า queue และดำเนินการต่อ

แผนภาพด้านล่าง illustrate queue lifecycle:

Queue Lifecycle | https://optimalbits.github.io/bull/

Job คืออะไร?

Job คือ data ใดๆ ที่ใช้บน queue ปกติเป็น JSON-like object

เหมือนรูปปกบทความ คุณคิดว่า job คือแต่ละคนบนคิวที่สนามบิน แต่ละคนถือกระเป๋า briefcase มีข้อมูลเฉพาะ และ instructions อื่นๆ (passport และ medical papers ถ้าจำเป็น) ช่วยตอนถึงคิวได้รับบริการ

คนใหม่เข้าคิวจะเข้าจากหลัง (เป็นคนสุดท้าย) และคนได้รับบริการจากหน้า นั่นคือ jobs processed เหมือนกัน แต่ละ job มี data ใช้สำหรับ processing ใหม่ jobs เพิ่มจากหลัง jobs ถูกนำออกจากหน้า

Job Producer คืออะไร?

Job producer คือ code ใดๆ ที่เพิ่ม job เข้า queue ในความเป็นจริงจะเป็น security guard สนามบิน ให้ directions บอกคนว่าคิวไหนสำหรับวัตถุประสงค์ต่างๆ

Job producer อาจมีอยู่ independent ของ job consumer หมายความว่าใน microservice setup service 某 อาจแค่เพิ่ม jobs เข้า queue แต่ไม่สนใจว่าจะ processed ยังไงหลังจากนั้น

Worker (Job Consumer) คืออะไร?

Worker หรือ job consumer คือ process หรือ function ที่ execute job คิดว่า worker คือ bank cashier รับบริการคนบนคิวที่ธนาคาร คนแรกเข้าคิวเป็นคนเดียว Cashier เรียกเขา และคิวว่าง

Cashier ขอรายละเอียดเฉพาะเพื่อ process transaction จากคนนั้น ขณะ cashier รับบริการคนนั้น อาจมีอีกสี่คนจับคิว พวกเขาอยู่คิวจน cashier เสร็จคนแรกก่อนเรียกคนถัดไป Process เดียวกับ queue workers — พวกเขา pick first job ใน queue และ process มัน

Failed Jobs คืออะไร?

บ่อยครั้ง jobs บางอัน fail ระหว่าง processing

เหตุผลที่ job อาจ fail:

• Invalid หรือ missing input data: เมื่อ data ที่ job ต้องการ process หายไป job จะ fail เช่น job ส่งอีเมลจะ fail ไม่มี recipient email
• Timeout: Job อาจถูก fail โดย queue mechanism หากใช้เวลานานกว่าปกติ อาจเนื่องจาก dependency issue หรืออย่างอื่น แตปกติไม่อยากให้ single job รันตลอดกาล
• Network หรือ infrastructure problems: ปัญหาเหล่านี้เกือบไม่อยู่ใน control ของคุณ แตเกิดขึ้นได้ Database connection error ตัวอย่าง fail job
• Dependency issues: บางครั้ง job ขึ้นอยู่กับ external resources เพื่อทำงานดี เมื่อ resources อื่นๆ unavailable หรือ unsuccessful job จะ fail

เมื่อ jobs fail คุณ config queue mechanism ให้ retry ได้ Retry ทันที หรือหลังเวลาที่คำนวณได้ ตั้ง maximum attempts ได้ แนะนำให้ตั้ง มิฉะนั้น job ที่ fail ตลอดจะรัน infinite

Queues มีประโยชน์สร้าง robust communication channels ระหว่าง microservices Multiple services ใช้ queue เดียวกันได้ Services ต่างๆ ทำหน้าที่ต่างกัน เมื่อ service เสร็จ task มัน push job ไป service อื่นที่มี workers รอ job นั้น service นั้น pick up และทำสิ่งที่ต้องกับ data

Queues ยัง offload heavy tasks จาก process ตามที่จะเห็นในบทความ time-consuming task เช่น ส่งอีเมล วางบน queue หลีกเลี่ยง slowing down response time

Queues ช่วยหลีกเลี่ยง single points of failure Process ที่อาจ fail และ retry ได้ ควร process 用 queue ที่ retry ได้หลังพักสักครู่

วิธีสร้าง Simple Application ที่ใช้ Queues

ในบทความนี้ เราสร้าง simple project 用 Node.js และ Redis ใช้ library Bull เพราะมันลดความซับซ้อนในการสร้าง queue system มาก โปรเจกต์จะมี single endpoint ส่ง emails

Create a New Node.js Project

mkdir nodejs-queue-project
cd nodejs-queue-project
npm init -y

คำสั่งข้างต้นสร้าง folder nodejs-queue-project และ package.json ไฟล์ ควรเป็นแบบนี้:

{
    "name": "nodejs-queue-project",
    "version": "1.0.0",
    "description": "",
    "main": "index.js",
    "scripts": {
        "test": "echo \"Error: no test specified\" && exit 1"
    },
    "keywords": [],
    "author": "",
    "license": "ISC"
}

Install the Required Dependencies

npm i express @types/express @types/node body-parser ts-node ts-lint typescript nodemon nodemailer @types/nodemailer

คำสั่งข้างต้นติดตั้ง packages และ dependencies ที่ต้องการ หลังติดตั้ง อัปเดต scripts section ของ package.json ให้มี dev command package.json ทั้งไฟล์ควรเป็นแบบนี้:

{
    "name": "nodejs-queue-project",
    "version": "1.0.0",
    "description": "",
    "main": "index.js",
    "scripts": {
        "dev": "nodemon src/app.ts"
    },
    "keywords": [],
    "author": "",
    "license": "ISC",
    "dependencies": {
        "@types/express": "^4.17.17",
        "@types/node": "^20.3.3",
        "@types/nodemailer": "^6.4.8",
        "body-parser": "^1.20.2",
        "express": "^4.18.2",
        "nodemailer": "^6.9.3",
        "nodemon": "^2.0.22",
        "ts-lint": "^4.5.1",
        "ts-node": "^10.9.1",
        "typescript": "^5.1.6"
    }
}

ไฟล์ข้างต้นแสดง dependencies ทั้งหมด คำสั่ง npm run dev จะรันเมื่อใช้ dev script

วิธีสร้าง Endpoint

อย่างแรก สร้าง folder src จะ contain code files ทั้งหมด ไฟล์แรกคือ root file ของ application — app.ts ตามใน package.json

เราใช้ app.ts import required packages และสร้าง simple server พร้อม single endpoint ส่ง email:

import express from "express";
import bodyParser from "body-parser";
import nodemailer from "nodemailer";
const app = express();
app.use(bodyParser.json());
app.post("/send-email", async (req, res) => {
    const { from, to, subject, text } = req.body;
    // Use a test account as this is a tutorial
    const testAccount = await nodemailer.createTestAccount();
    const transporter = nodemailer.createTransport({
        host: "smtp.ethereal.email",
        port: 587,
        secure: false,
        auth: {
            user: testAccount.user,
            pass: testAccount.pass,
        },
        tls: {
            rejectUnauthorized: false,
        },
    });
    console.log("Sending mail to %s", to);
    let info = await transporter.sendMail({
        from,
        to,
        subject,
        text,
        html: `${text}`,
    });
    console.log("Message sent: %s", info.messageId);
    console.log("Preview URL: %s", nodemailer.getTestMessageUrl(info));
    res.json({ message: "Email Sent", });
});
app.listen(4300, () => {
    console.log("Server started at https://localhost:4300");
});

ตอนนี้ start server โดยรัน npm run dev ใน terminal ควรเห็นข้อความ Server started at [https://localhost:4300](https://localhost:4300)

npm run dev message

ทดสอบ endpoint ด้วย Postman:

Endpoint testing on Postman

Request ใช้เวลาเกือบ 4 วินาที ตาม screenshot นี่ช้ามากสำหรับ endpoint หากดู terminal ควรเห็น URL preview email ที่ส่ง

เปิด link ดู email long อย่างไร

Email content

วิธีสร้าง Queue

เพื่อให้ process เร็วขึ้น อีเมลสามารถ queue ให้ส่งภายหลัง และส่ง response ให้ user ทันที

ติดตั้ง library bull และ @types/bull เราจะใช้สร้าง queue:

npm i bull @types/bull

สร้าง queue ใหม่用 bull ง่ายแค่ instantiate Bull object พร้อมชื่อ queue:

// This goes at the top of your file
import Bull from 'bull';
const emailQueue = new Bull("email");

เมื่อสร้าง queue ด้วยแค่ชื่อ มันพยายามใช้ default Redis connection URL: localhost:6379 หากชอบ URL อื่น ส่ง object ที่สองให้ Bull class เป็น options:

const emailQueue = new Bull("email", { redis: "localhost:6379", });

ตอนนี้ สร้าง simple function เป็น job producer เพิ่ม job เข้า queue ทุกครั้งที่ request เข้ามา:

type EmailType = {
    from: string;
    to: string;
    subject: string;
    text: string;
};
const sendNewEmail = async (email: EmailType) => {
    emailQueue.add({ ...email });
};

Function sendNewEmail ใหม่นี้รับ object รายละเอียดอีเมลใหม่ type EmailType มี sender email (from), recipient (to), subject, และ content (text) แล้ว push job ใหม่เข้า queue

app.post("/send-email", async (req, res) => {
    const { from, to, subject, text } = req.body;
    await sendNewEmail({ from, to, subject, text });
    console.log("Added to queue");
    res.json({ message: "Email Sent", });
});

const processEmailQueue = async (job: Job) => {
    // Use a test account as this is a tutorial
    const testAccount = await nodemailer.createTestAccount();
    const transporter = nodemailer.createTransport({
        host: "smtp.ethereal.email",
        port: 587,
        secure: false,
        auth: {
            user: testAccount.user,
            pass: testAccount.pass,
        },
        tls: {
            rejectUnauthorized: false,
        },
    });
    const { from, to, subject, text } = job.data;
    console.log("Sending mail to %s", to);
    let info = await transporter.sendMail({
        from,
        to,
        subject,
        text,
        html: `${text}`,
    });
    console.log("Message sent: %s", info.messageId);
    console.log("Preview URL: %s", nodemailer.getTestMessageUrl(info));
    return nodemailer.getTestMessageUrl(info);
};

LRANGE bull:email:wait 0 -1

emailQueue.process(processEmailQueue);

import express from "express";
import bodyParser from "body-parser";
import nodemailer from "nodemailer";
import Bull, { Job } from "bull";
const app = express();
app.use(bodyParser.json());
const emailQueue = new Bull("email", { redis: "localhost:6379", });
type EmailType = {
    from: string;
    to: string;
    subject: string;
    text: string;
};
const sendNewEmail = async (email: EmailType) => {
    emailQueue.add({ ...email });
};
const processEmailQueue = async (job: Job) => {
    // Use a test account as this is a tutorial
    const testAccount = await nodemailer.createTestAccount();
    const transporter = nodemailer.createTransport({
        host: "smtp.ethereal.email",
        port: 587,
        secure: false,
        auth: {
            user: testAccount.user,
            pass: testAccount.pass,
        },
        tls: {
            rejectUnauthorized: false,
        },
    });
    const { from, to, subject, text } = job.data;
    console.log("Sending mail to %s", to);
    let info = await transporter.sendMail({
        from,
        to,
        subject,
        text,
        html: `${text}`,
    });
    console.log("Message sent: %s", info.messageId);
    console.log("Preview URL: %s", nodemailer.getTestMessageUrl(info));
};
emailQueue.process(processEmailQueue);
app.post("/send-email", async (req, res) => {
    const { from, to, subject, text } = req.body;
    await sendNewEmail({ from, to, subject, text });
    console.log("Added to queue");
    res.json({ message: "Email Sent", });
});
app.listen(4300, () => {
    console.log("Server started at https://localhost:4300");
});