Computer >> Hướng Dẫn Máy Tính >  >> Lập Trình >> Redis

Xây Dựng Custom Message Queue Với Redis Lists & TypeScript

Giới Thiệu

Bạn từng thử tạo Message Queue riêng nhưng gặp khó khăn? Bạn không đơn độc. Trong tutorial này, chúng ta sẽ xây dựng message queue từ con số 0 dùng Redis lists. Có nhiều phương pháp xây message queue với Redis (streams, lists, pub/sub), nhưng chúng ta tập trung vào cách đơn giản, trực quan nhất: lists.

Công Cụ Sử Dụng

  • Upstash
  • Đôi tay

Yêu Cầu

  • Bun
  • Đôi tay

Thiết Lập Upstash Redis

Tạo Redis instance: vào Upstash, nhấn Create Database. Cuộn xuống lấy connection string để connect client.

Ví Dụ Connection String:

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

Khởi Động Dự Án

Khởi tạo TypeScript project với Bun. Chọn Bun không chỉ vì nhanh hơn Node — setup cũng dễ hơn. Và thật ấn tượng về tốc độ! 🚀

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

Cấu Trúc Dự Án

 ┣ 📂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 của Queue:

Xây Dựng Custom Message Queue Với Redis Lists & TypeScript

Job Class

Job class cần theo dõi status mỗi job (để quyết định process, retry, hoặc move nếu done). Mỗi job có ID và data (generic cho UX tốt). Cần link job đến queue và queue name để quản lý dễ dàng.

type OwnerQueue = { redis: Redis; queueName: string; };
export type JobStatuses = | "created" | "waiting" | "active" | "succeeded" | "failed";

export class Job<T> {
  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, làm Job generic. Có thể tạo Redis instance riêng cho mỗi Job element nhưng quản lý phức tạp. Cách của chúng ta cho phép config Redis instance trong queue và pass instance khi cần. Tương tự cho queueName. Để save job vào queue, cần: Lua script tương tác Redis và 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 = <T>(data: T, status: JobStatuses): string => {
  return JSON.stringify({ data, status });
};

formatMessageQueueKey concat string. convertToJSONString serialize data generic để lưu Redis (không thể pass JS object trực tiếp).

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

Thay vì 3 gọi Redis riêng biệt (hexists, hset, lpush), gộp thành 1 call để giảm round trips.

Thêm Method save()

private createQueueKey(key: string) {
  return formatMessageQueueKey(this.config.queueName, key);
}
async save(): Promise<string | null> {
  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 chạy Lua script. Tham số: script, số lượng keys, các key, các argument.

Thêm Methods fromId & fromData

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

Các method này quan trọng khi process job: chỉ có jobId, cần reconstruct Job từ đầu.

Queue Class

Mục tiêu:

  • Queue retain hoặc remove data khi success/failure (có thể reprocess sau).
  • Thiết kế cho concurrency — multiple jobs chạy đồng thời.
  • Cho phép pass callback function process data, infer type Job cho DX tốt.
  • Cho phép gọi Job.save() từ Queue, pass Redis instance và queueName.
  • Cho phép destroy queue và remove job khỏi queue.

Định Nghĩa 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 config nhận info external: queue name, Redis instance, setting retain/remove. Chấp nhận bất kỳ Redis implementation nào (đặc biệt thích Upstash 😌). Kế thừa EventEmitter để notify khi có sự kiện.

Ví Dụ Khởi Tạo

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

Add Job

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

Job lấy config parent queue và payload (data lưu Redis). Gọi save().

Processing

Phần thách thức nhất. User chỉ định concurrency và worker — callback process job infer type job. Cần track running/queued jobs để pick next job an toàn.

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

Generic TJobPayload giúp TypeScript infer type cho intellisense tốt hơn.

jobTick()

Queue FIFO: pop job từ right queue → run worker → emit result → gọi jobTick() cho job tiếp theo.

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;
  }
}

Dùng blocking call + lpush giảm round trips. Blocking call ngăn workers khác process cùng job (tránh race condition). Di chuyển job từ 'waiting' sang 'active'.

executeJob()

private async executeJob<TJobPayload>(jobCreatedById: Job<TJobPayload>) {
  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<TJobPayload>(jobCreatedById, hasError);
    this.emit(jobStatus, job.id);
    return;
  }
}

Pass job data cho worker. Thành công: queued+1, running-1. Lỗi: set hasError. Gọi finishJob emit status với jobId.

User listen events:

queue.on("succeeded", (jobId) => console.log("Succeeded jobId", jobId));

finishJob()

private async finishJob<TJobPayload>(
  job: Job<TJobPayload>,
  hasFailed?: boolean
): Promise<[JobStatuses, Job<TJobPayload>]> {
  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];
}

Dùng multi() giảm round trips. Redis defer execution đến exec(). Nếu keepOnFailure/Success: tạo 2 set — data job và list job IDs. Adjust status. Exec multi, return status và 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)

Demo Toàn Bộ

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<Payload>((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}`);
  }
}

Thử Thách Mở Rộng

  • Implement retry logic với exponential backoff cho Redis access và worker processes.
  • Phát triển cơ chế 'at least once' guarantee.
  • Thử execute workers trong Service Workers để performance tốt hơn.
  • Thêm scheduled jobs.

Kết Luận

Cách tốt nhất học là tự xây dựng, còn tốt hơn là dùng Upstash Redis. Keep rocking.

🔗 Project's Github address