สร้าง Custom Message Queue ด้วย Redis Lists และ TypeScript

เคยพยายามสร้าง Message Queue เองแต่เจอปัญหา? คุณไม่ได้อยู่คนเดียว ในบทเรียนนี้ เราจะสร้าง message queue จากศูนย์ด้วย Redis lists แม้จะมีหลายวิธีสร้าง message queues ด้วย Redis เช่น streams, lists, pub/subs เราจะโฟกัสที่วิธีที่ง่ายและตรงไปตรงมาที่สุด: lists มาสำร่วจกันในคู่มือปฏิบัติจริงนี้

สิ่งที่เราจะใช้

Upstash
คู่มือ

สิ่งที่คุณต้องการ

Bun
คู่มือ

ตั้งค่า Upstash Redis

ก่อนอื่น ตั้งค่า Redis instance ไปที่ Upstash กด Create Database เลื่อนลงหา connection string ที่ใช้เชื่อม client

ตัวอย่าง connection string:

redis://XXXXe@social-XXX-39281.upstash.io:39281

เริ่มโปรเจกต์

เริ่ม TypeScript project ด้วย Bun ไม่ใช่แค่เร็วกว่า Node แต่ง่ายต่อการตั้งค่ามากกว่า และเร็วมาก! 🚀

mkdir upstash-mq
cd upstash-mq
bun init
> package name (upstash-mq-tutorial): upstash-mq
> entry point (index.ts):
> Done!
bun add ioredis

โครงสร้างโปรเจกต์

 ┣ 📂src
 ┃ ┣ 📂lua-scripts
 ┃ ┃ ┣ 📜add-job.lua
 ┃ ┃ ┗ 📜remove-job.lua
 ┃ ┣ 📜index.ts
 ┃ ┣ 📜job.ts
 ┃ ┣ 📜queue.ts
 ┃ ┗ 📜utils.ts
 ┣ 📜.env
 ┣ 📜.gitignore
 ┣ 📜README.md
 ┣ 📜bun.lockb
 ┣ 📜index.ts
 ┣ 📜package.json
 ┗ 📜tsconfig.json

Visual representation ของ Queue:

สร้าง Custom Message Queue ด้วย Redis Lists และ TypeScript

Job

Job class ต้องการสิ่งสำคัญๆ: ติดตามสถานะแต่ละ job (process, retry, หรือ move ถ้าเสร็จแล้ว) แต่ละ job มี ID และ data (generic สำหรับ UX ที่ดี) และ link ไปยัง queue พร้อม queue name

โครงสร้างหลักของ Job class:

type OwnerQueue = { redis: Redis; queueName: string; };
export type JobStatuses = | "created" | "waiting" | "active" | "succeeded" | "failed";
export class Job {
  id: string;
  status: JobStatuses;
  config: OwnerQueue;
  data: T;
  constructor(ownerConfig: OwnerQueue, data: T, jobId = randomUUID()) {
    this.id = jobId;
    this.status = "created";
    this.data = data;
    this.config = ownerConfig;
  }
}

ทำ data ให้ generic โดยทำ Job ให้ generic ส่วนที่เหลือตรงไปตรงมา ไม่ควรสร้าง Redis instance แยกสำหรับแต่ละ Job (ซับซ้อน) ใช้แนวทางของเรา: config Redis instance ใน queue แล้วส่งต่อไปได้ queueName เช่นกัน

เพื่อ save jobs เข้า queues ต้องการ 2 อย่าง: Lua script สำหรับโต้ตอบ Redis และ utilities

Utilities

import { JobStatuses } from "./job";
const MQ_PREFIX = "UpstashMQ";
export const formatMessageQueueKey = (queueName: string, key: string) => {
  return `${MQ_PREFIX}:${queueName}:${key}`;
};
export const convertToJSONString = (data: T, status: JobStatuses): string => {
  return JSON.stringify({ data, status });
};

formatMessageQueueKey รวม string ให้ convertToJSONString serialize data generic

Lua Script: add-job.lua

--[[ key 1 -> [prefix]:name:jobs key 2 -> [prefix]:name:waiting arg 1 -> job id arg 2 -> job data ]]
local jobId = ARGV[1]
local payload = ARGV[2]
if redis.call("hexists", KEYS[1], jobId) == 1 then return nil end
redis.call("hset", KEYS[1], jobId, payload)
redis.call("lpush", KEYS[2], jobId)
return jobId

แทน 3 เรียกแยก (hexists, hset, lpush) รวมเป็น 1 call ลด round trips

เพิ่มเมธอด `save()`

private createQueueKey(key: string) {
  return formatMessageQueueKey(this.config.queueName, key);
}
async save(): Promise {
  const addJobToQueueScript = await Bun.file("./src/lua-scripts/add-job.lua").text();
  const resJobId = (await this.config.redis.eval(
    addJobToQueueScript,
    2,
    this.createQueueKey("jobs"),
    this.createQueueKey("waiting"),
    this.id,
    convertToJSONString(this.data, this.status)
  )) as string | null;
  if (resJobId) {
    this.id = resJobId;
    return resJobId;
  }
  return null;
}

redis.eval รัน Lua script: param 1=script, 2=จำนวน keys, 3-4=keys, ต่อไป=arguments

เมธอดเพิ่มเติมสำหรับอนาคต

fromId = async (jobId: string): Promise | null> => {
  const jobData = await this.config.redis.hget(this.createQueueKey("jobs"), jobId);
  if (jobData) { return this.fromData(jobId, jobData); }
  return null;
};
private fromData = (jobId: string, stringifiedJobData: string): Job => {
  const parsedData = JSON.parse(stringifiedJobData) as Job;
  const job = new Job(this.config, parsedData.data, jobId);
  job.status = parsedData.status;
  return job;
};

fromId ดึง job data จาก Redis สร้าง Job instance คืนค่า เพื่อให้ queue process ต่อ

Queue

วัตถุประสงค์:

Queue retain หรือ remove data ตาม success/failure (อาจต้อง reprocess)
ออกแบบสำหรับ concurrency (หลาย jobs พร้อมกัน)
รับ callback function สำหรับ process data (infer job type สำหรับ DX)
เรียก Job.save() จากใน Queue (ส่ง Redis instance และ queueName)
Destroy queue และ remove job ได้

กำหนด Queue

export type QueueConfig = {
  redis: Redis;
  queueName: string;
  keepOnSuccess?: boolean;
  keepOnFailure?: boolean;
};
export class Queue extends EventEmitter {
  config: QueueConfig;
  concurrency = 0;
  worker: any;
  running = 0;
  queued = 0;
  constructor(config: QueueConfig) {
    super();
    this.config = {
      redis: config.redis,
      queueName: config.queueName,
      keepOnFailure: config.keepOnFailure ?? true,
      keepOnSuccess: config.keepOnSuccess ?? true,
    };
  }
  createQueueKey(key: string) {
    return formatMessageQueueKey(this.config.queueName, key);
  }
}

Queue class มี config รับข้อมูลภายนอก (queue name, Redis instance, keep settings) ยอมรับ Redis implementation ใดๆ (ชอบ Upstash 😌) Extends EventEmitter เพื่อ notify เมื่อมี event

ตัวอย่าง initialization:

const queue = new Queue({
  redis: new Redis(process.env.UPSTASH_REDIS_URL),
  queueName: "upstash-rocks",
  keepOnFailure: true,
  keepOnSuccess: true,
});

Add

async add(payload: T) {
  return new Job(this.config, payload).save();
}

Job รับ config ของ parent queue และ payload (data ที่จะเก็บใน Redis) แล้ว save

Processing

ส่วนที่ท้าทายที่สุด: ต้องการ concurrency count และ worker (callback function ที่ infer job type) และ mechanism ติดตาม running/queued jobs

async process(
  worker: (job: TJobPayload) => void,
  concurrency: number
): Promise {
  this.concurrency = concurrency;
  this.worker = worker;
  this.running = 0;
  this.queued = 1;
  this.jobTick();
}

Generic TJobPayload เพื่อ DX ให้ users ได้ intellisense

กระบวนการ jobTick():

Pop job จากขวาของ queue (FIFO)
Run worker function
Emit result
Call jobTick() อีกครั้ง

private jobTick() {
  this.getNextJob()
    .then(async (jobId) => {
      this.running += 1;
      this.queued -= 1;
      if (this.running + this.queued < this.concurrency) {
        this.queued += 1;
        setImmediate(this.jobTick);
      }
      if (!jobId) { return; }
      const jobCreatedById = await new Job(this.config, null).fromId(jobId);
      if (jobCreatedById) {
        await this.executeJob(jobCreatedById);
      } else {
        console.error(`Job not found with ID: ${jobId}`);
      }
    })
    .catch((error) => { console.error("Error in jobTick:", error); })
    .finally(() => { setImmediate(() => this.jobTick()); });
}

getNextJob()

private async getNextJob() {
  try {
    const jobId = await this.config.redis.brpoplpush(
      this.createQueueKey("waiting"),
      this.createQueueKey("active"),
      0
    );
    return jobId;
  } catch (error) {
    console.error("Error fetching the next job:", error);
    throw error;
  }
}

ใช้ blocking call + lpush ลด round trips ป้องกัน race conditions ย้าย jobs waiting → active

executeJob

private async executeJob(jobCreatedById: Job) {
  let hasError = false;
  try {
    await this.worker(jobCreatedById.data);
    this.running -= 1;
    this.queued += 1;
  } catch (error) {
    hasError = true;
  } finally {
    const [jobStatus, job] = await this.finishJob(jobCreatedById, hasError);
    this.emit(jobStatus, job.id);
    return;
  }
}

ส่ง job data ให้ worker สำเร็จ: queued+1, running-1 (สำคัญสำหรับ concurrency) Error: set hasError flag เรียก finishJob emit status

finishJob

private async finishJob(
  job: Job,
  hasFailed?: boolean
): Promise<[JobStatuses, Job]> {
  const multi = this.config.redis.multi();
  multi.lrem(this.createQueueKey("active"), 0, job.id);
  if (hasFailed) {
    if (this.config.keepOnFailure) {
      multi.hset(this.createQueueKey("jobs"), job.id, convertToJSONString(job.data, job.status));
      multi.sadd(this.createQueueKey("failed"), job.id);
    } else {
      multi.hdel(this.createQueueKey("jobs"), job.id);
    }
    job.status = "failed";
  } else {
    if (this.config.keepOnSuccess) {
      multi.hset(this.createQueueKey("jobs"), job.id, convertToJSONString(job.data, job.status));
      multi.sadd(this.createQueueKey("succeeded"), job.id);
    } else {
      multi.hdel(this.createQueueKey("jobs"), job.id);
    }
    job.status = "succeeded";
  }
  await multi.exec();
  return [job.status, job];
}

ใช้ multi() ลด round trips (defer จน exec()) keepOnFailure/Success: สร้าง 2 sets (job data + job IDs) สำหรับทั้ง success/failure ปรับ status คืนค่า status และ job

removeJob และ destroy

async removeJob(jobId: string) {
  const addJobToQueueScript = await Bun.file("./src/lua-scripts/remove-job.lua").text();
  return await this.config.redis.eval(
    addJobToQueueScript,
    5,
    this.createQueueKey("succeeded"),
    this.createQueueKey("failed"),
    this.createQueueKey("waiting"),
    this.createQueueKey("active"),
    this.createQueueKey("jobs"),
    jobId
  );
}
async destroy() {
  const args = ["id", "jobs", "waiting", "active", "succeeded", "failed"].map(key => this.createQueueKey(key));
  const res = await this.config.redis.del(...args);
  return res;
}

remove-job.lua

--[[ key 1 -> [prefix]:test:succeeded key 2 -> [prefix]:test:failed key 3 -> [prefix]:test:waiting key 4 -> [prefix]:test:active key 5 -> [prefix]:test:jobs arg 1 -> jobId ]]
local jobId = ARGV[1]
if (redis.call("sismember", KEYS[1], jobId) + redis.call("sismember", KEYS[2], jobId)) == 0 then
  redis.call("lrem", KEYS[3], 0, jobId)
  redis.call("lrem", KEYS[4], 0, jobId)
end
redis.call("srem", KEYS[1], jobId)
redis.call("srem", KEYS[2], jobId)
redis.call("hdel", KEYS[5], jobId)

ดูการทำงานจริง

import { sleep } from "bun";
import Redis from "ioredis";
import { Queue } from "./queue";

type Payload = { id: number; data: string; };

const queue = new Queue({
  redis: new Redis(process.env.UPSTASH_REDIS_URL),
  queueName: "mytest-queue",
});

async function main() {
  await generateQueueItems(queue, 20);
  console.log("Sleep starting for 5 sec");
  await sleep(5000);
  queue.on("succeeded", (jobId) => console.log("Succeeded jobId", jobId));
  await queue.process((job) => {
    console.log("Processing job:", job.data);
    sleep(1000);
  }, 3);
}
main();

async function generateQueueItems(queue: Queue, itemCount: number) {
  for (let i = 0; i < itemCount; i++) {
    const payload = { id: i, data: `dummy-data-${i}` };
    const jobId = await queue.add(payload);
    console.log(`Added item ${i} with jobId: ${jobId}`);
  }
}

Bonus Challenges

Implement retry logic with exponential backoffs สำหรับ Redis access และ worker processes
Develop 'at least once' guarantee mechanism
Try to execute workers in Service Workers เพื่อ performance
Add scheduled jobs

Wrap up

วิธีเรียนรู้ที่ดีที่สุดคือการสร้างด้วยตัวเอง และดีกว่านั้นคือทำด้วย Upstash Redis Keep rocking.

🔗 Project's Github address

สร้าง Custom Message Queue ด้วย Redis Lists และ TypeScript

สิ่งที่เราจะใช้

สิ่งที่คุณต้องการ

ตั้งค่า Upstash Redis

ตัวอย่าง connection string:

เริ่มโปรเจกต์

โครงสร้างโปรเจกต์

Job

Utilities

Lua Script: add-job.lua

เพิ่มเมธอด save()

เมธอดเพิ่มเติมสำหรับอนาคต

Queue

กำหนด Queue

ตัวอย่าง initialization:

Add

Processing

getNextJob()

executeJob

finishJob

removeJob และ destroy

remove-job.lua

ดูการทำงานจริง

Bonus Challenges

Wrap up

เพิ่มเมธอด `save()`