আপনি কি কখনো নিজে মেসেজ কিউ তৈরি করতে চেয়েছেন কিন্তু চ্যালেঞ্জ মেরে পड़ेেছেন? আপনি একলা নন। এই টিউটোরিয়ালে আমরা Redis lists ব্যবহার করে স্ক্র্যাচ থেকে একটি মেসেজ কিউ বিল্ড করব। Redis-এ মেসেজ কিউ তৈরির অনেক পদ্ধতি আছে (streams, lists, pub/subs), কিন্তু আমরা সবচেয়ে সিম্পল এবং সিদা পদ্ধতিঃ lists-এ ফোকাস করব।
আমরা কী ব্যবহার করব
- Upstash
- হাতের যুগল
আপনার কী প্রয়োজন
- Bun
- হাতের যুগল
Upstash Redis সেটআপ করা
প্রথমে একটি Redis ইনস্ট্যান্স সেটআপ করুন। Upstash-এ যান, Create Database ক্লিক করুন। এরপর কানেকশন স্ট্রিং (connection string) খুঁজে বের করুন, যা ক্লায়েন্ট কানেক্ট করতে ব্যবহার করব।
উদাহরণ কানেকশন স্ট্রিং:
redis://XXXXe@social-XXX-39281.upstash.io:39281প্রজেক্ট শুরু করা
Bun ব্যবহার করে TypeScript প্রজেক্ট শুরু করুন। Bun শুধু ফাস্টার নয়, সেটআপও সহজ।
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আমাদের Queue-র ভিজ্যুয়াল রিপ্রেজেন্টেশন:
Job
আমাদের Job ক্লাসের কিছু কি জিনিস প্রয়োজন। প্রথমে প্রতিটি জবের স্ট্যাটাস ট্র্যাক করতে হবে। এটি আমাদের সিদ্ধান্ত নিতে সাহায্য করে কি প্রসেস করব, রি-ট্রাই করব বা অলরেডি ফিনিশড হলে অন্য জায়গায় মুভ করব। প্রতিটি জবের একটি ID এবং কিছু ডেটা থাকে, যা জেনেরিক হতে হবে যাতে ইউজার এক্সপেরিয়েন্স ভালো হয়। অবশেষে প্রতিটি জবকে এর কিউ এবং কিউর নামের সাথে লিংক করতে হবে সহজ ম্যানেজমেন্টের জন্য।
Job ক্লাসের ব্যাকবোন:
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 জেনেরিক করতে হলে Job-কেই জেনেরিক করতে হয়। বাকি সহজ। প্রতিটি Job এলিমেন্টের জন্য আলাদা Redis ইনস্ট্যান্স তৈরি করা যায় কিন্তু ম্যানেজ করা জটিল। আমাদের পদ্ধতিতে Redis ইনস্ট্যান্স কিউতে কনফিগার করা সহজ, এবং প্রয়োজনে পাস করা যায়। queueName-ও একই। যেহেতু কিউতে জব সেভ করতে বারবার কিউ নাম লাগবে, জবগুলো তার পেরেন্ট কিউ সম্পর্কে জানতে হবে। কিউতে জব সেভ করতে দুটি জিনিস লাগে: Redis-এ ইন্টারঅ্যাক্ট করার লুয়া স্ক্রিপ্ট এবং কিছু ইউটিলিটি।
প্রথমে ইউটিলিটি তৈরি করুন:
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 ইউটিলিটি স্ট্রিং কনক্যাটেনেট করে। ডেটা সিরিয়ালাইজ করার জন্য জেনেরিক ফাংশন convertToJSONString বানানো হয়েছে।
এখন প্রথম লুয়া স্ক্রিপ্ট যোগ করুন:
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আমরা আলাদা আলাদা Redis কলও করতে পারতাম (redis.hexists, redis.hset, redis.lpush), কিন্তু তখন তিনটি আলাদা কল লাগত। রাউন্ড ট্রিপ কমাতে পুরো প্রসেস এক কলে করার জন্য লুয়া স্ক্রিপ্ট ব্যবহার করা হয়েছে।
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 দিয়ে লুয়া স্ক্রিপ্ট এক্সিকিউট করা হয়। প্যারামিটারগুলো: স্ক্রিপ্ট, আর্গুমেন্ট সংখ্যা, কি-গুলো, এবং আসল আর্গুমেন্ট।
Job ক্লাসে আরও কিছু মেথড যোগ করুন ভবিষ্যতের জন্য:
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; };এই ফাংশনগুলো জব প্রসেস করার সময় জরুরি হবে। তখন শুধু জব আইডি থাকবে, তখন থেকে জব রিকনস্ট্রাক্ট করতে fromId ব্যবহার করা হবে।
Queue-এ যাওয়া যাক
save() পার্ট শেষ হলে Queue-র ডিটেইলস দেখি। আমাদের লক্ষ্য:
- সাকসেস বা ফেইল্যুরে ডেটা রিটেইন বা রিমুভ করা (রিপ্রসেসিংয়ের জন্য)।
- কoncesurrency সাপোর্ট (একসাথে একাধিক জব রান)।
- কলোক ফাংশন পাস করা (জবের টাইপ ইনফার করা)।
- Queue থেকে
Job.save()কল করা (Redis ইনস্ট্যান্স এবং queueName পাস করে)। - Queue ডেস্ট্রয় করা এবং জব রিমুভ করা।
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 ক্লাসে কনফিগ আছে (queueName, Redis ইনস্ট্যান্স, রিটেইন সেটিংস)। যেকোনো Redis ইমপ্লিমেন্টেশন এক্সেপ্ট করা হয় (Upstash পছন্দ)। EventEmitter এক্সটেন্ড করা হয়েছে ইভেন্ট নোটিফিকেশনের জন্য।
উদাহরণ ইনিশিয়ালাইজেশন:
const queue = new Queue({ redis: new Redis(process.env.UPSTASH_REDIS_URL), queueName: "upstash-rocks", keepOnFailure: true, keepOnSuccess: true, });Add
async add<T>(payload: T) { return new Job<T>(this.config, payload).save(); }জব পেরেন্ট কিউর কনফিগ এবং পেলোড নেয়, তারপর সেভ করে।
এখন:
const queue = new Queue({ redis: new Redis(process.env.UPSTASH_REDIS_URL!), queueName: "mytest-queue", keepOnFailure: true, keepOnSuccess: true, }); const payload = { upstash: "best-redis-ever", }; await queue.add(payload);এখন প্রসেস করার উপায় লাগবে।
Processing
এটি কিউ ইমপ্লিমেন্টেশনের সবচেয়ে চ্যালেঞ্জিং পার্ট। ইউজার কনকারেন্সি সংখ্যা এবং একটি ওয়ার্কার (কলব্যাক ফাংশন) স্পেসিফাই করবে। রানিং এবং কিউড জব ট্র্যাক করতে হবে যাতে নেক্সট জব সেফলি পিক করা যায়।
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(); }জেনেরিক TJobPayload গ্রহণ করার উদ্দেশ্য ডেভেলপার এক্সপেরিয়েন্স উন্নত করা (ইন্টেলিসেন্স)।
jobTick()-এ যাওয়ার আগে প্রসেস চিন্তা করুন:
- FIFO বেসিসে কিউর রাইট থেকে জব পপ করা।
- ওয়ার্কার ফাংশন রান করা।
- জব কমপ্লিট হলে রেজাল্ট ইমিট করা।
- আবার
jobTick()কল করে নেক্সট জব প্রসেস করা।
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; } }ব্লকিং কল সহ lpush ব্যবহার করে রাউন্ড ট্রিপ কমানো হয়েছে। ব্লকিং কল দিয়ে রেস কন্ডিশন এড়ানো হয়। জব 'waiting' থেকে 'active'-এ মুভ করা হয়।
জব আইডি পাওয়ার পর রানিং কাউন্ট বাড়ানো, কিউড কাউন্ট কমানো, এবং কoncesurrency লিমিট মেনে যতগুলো সম্ভব নেওয়া জব আরম্ভ করা হয়।
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; } }জবের ডেটা ওয়ার্কার-এ পাস করা হয়। সাকসেস হলে কিউড বাড়ানো, রানিং কমানো। এরর হলে hasError ফ্ল্যাগ সেট করা হয়। শেষে finishJob কল করে স্ট্যাটাস ইমিট করা হয়।
সাইড নোট: ইউজার ইভেন্ট লিসেন করতে পারেন:
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]; }multi() ব্যবহার করে রাউন্ড ট্রিপ মিনিমাইজ করা হয়েছে। keepOnFailure এবং keepOnSuccess অনুযায়ী সেট তৈরি করা হয়। জবের স্ট্যাটাস আপডেট করে exec এক্সিকিউট করা হয়।
বাকি দুটি মেথড (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) destroy() পুরো কিউ וাইপ করে, অন্যটি স্পেসিফিক জব ডিলিট করে।
সব কিছু একশনে দেখি
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}`); } }বোনাস চ্যালেঞ্জ
- রিডিস অ্যাক্সেস এবং ওয়ার্কার প্রসেসের জন্য এক্সপোনেনশিয়াল ব্যাকঅফ সহ রি-ট্রাই লজিক বাস্তবায়ন করুন।
- 'অ্যাট লিস্ট ওন্স' গ্যারান্টি মেকানিজম ডেভেলপ করুন।
- সার্ভিস ওয়ার্কার-এ ওয়ার্কার এক্সিকিউট করার চেষ্টা করুন বেটার পারফরম্যান্সের জন্য।
- শিডিউলড জব যোগ করুন।
র্যাপ আপ
কিছু শিখার সেরা উপায় হলো সেটি বিল্ড করা, আর আরও ভালো উপায় হলো Upstash Redis ব্যবহার করে বিল্ড করা। রকিং থাকুন।
🔗 প্রজেক্টের GitHub ঠিকানা
