پست

طراحی یک Distributed Job Scheduler مشابه Airflow

بررسی گام‌به‌گام طراحی یک زمان‌بند توزیع‌شده برای اجرای ۱۰ هزار شغل در ثانیه با دقت ۲ ثانیه

طراحی یک Distributed Job Scheduler مشابه Airflow

درک مسئله

Job Scheduler برنامه‌ای است که کارها را به‌صورت خودکار در زمان‌ها یا بازه‌های مشخص زمان‌بندی و اجرا می‌کند. از آن برای خودکارسازی کارهای تکراری، نگهداری برنامه‌ریزی‌شده و اجرای پردازش‌های دسته‌ای استفاده می‌شود.

پیش از شروع، دو اصطلاح کلیدی را روشن می‌کنیم:

  • Task (وظیفه): مفهوم انتزاعی کاری که باید انجام شود، مثلاً «ارسال ایمیل». یک Task قابل استفاده مجدد است و Jobهای مختلف می‌توانند آن را چندین بار اجرا کنند.
  • Job (شغل): یک نمونه از Task که از سه بخش تشکیل می‌شود: Task مورد نظر، زمان‌بندی اجرا و پارامترهای لازم. مثلاً «ارسال ایمیل به [email protected] ساعت ۱۰ صبح جمعه».

مسئولیت اصلی Scheduler این است که مجموعه‌ای از Jobها را بگیرد و طبق برنامه اجرا کند.

نیازمندی‌های عملکردی

نیازمندی‌های اصلی:

  • کاربر بتواند Jobهایی را زمان‌بندی کند که فوراً، در یک تاریخ مشخص در آینده، یا به‌صورت تکرارشونده (مثلاً «هر روز ساعت ۱۰ صبح») اجرا شوند.
  • کاربر بتواند وضعیت Jobهای خود را پایش کند.

خارج از محدوده:

  • لغو یا زمان‌بندی مجدد Job.

نیازمندی‌های غیرعملکردی

این لحظه مناسبی است که در مصاحبه درباره مقیاس سیستم سؤال کنید. اگر من مصاحبه‌کننده بودم، می‌گفتم سیستم باید ۱۰ هزار Job در ثانیه را اجرا کند.

نیازمندی‌های اصلی:

  • سیستم باید Highly Available باشد (اولویت با Availability است نه Consistency).
  • هر Job باید حداکثر ۲ ثانیه بعد از زمان برنامه‌ریزی‌شده اجرا شود.
  • سیستم باید Scalable باشد و تا ۱۰ هزار Job در ثانیه را پشتیبانی کند.
  • اجرای Job باید به‌صورت At-Least-Once تضمین شود.

خارج از محدوده:

  • اعمال سیاست‌های امنیتی.
  • داشتن خط لوله CI/CD.

رویکرد برنامه‌ریزی

این سؤال بیشتر روی پردازش داده تمرکز دارد تا تجربه کاربری. بنابراین چارچوب تحویل را دنبال می‌کنیم: اول Entityهای اصلی، بعد API، بعد جریان داده، و در نهایت طراحی سطح بالا و Deep Dive.

تعریف Entityهای اصلی

حتی در سؤال‌هایی که بیشتر درباره زیرساخت هستند، بهتر است با یک فهرست ساده از Entityها شروع کنید. لازم نیست همین حالا همه ستون‌ها و جزئیات را بدانید. بعد از اینکه تصویر روشن‌تری از سیستم پیدا کردیم، به فیلدها می‌پردازیم. فقط برنامه‌تان را به مصاحبه‌کننده بگویید تا هر دو در یک مسیر باشید. من معمولاً می‌گویم فعلاً با یک فهرست ساده شروع می‌کنم و مدل داده را در طراحی سطح بالا کامل می‌نویسم.

برای پوشش نیازمندی‌های عملکردی به این Entityها نیاز داریم:

  • Task: کاری که باید اجرا شود.
  • Job: نمونه‌ای از یک Task که باید در زمان مشخص و با مجموعه‌ای از پارامترها اجرا شود.
  • Schedule: زمان‌بندی اجرا، که یا یک عبارت CRON است یا یک DateTime مشخص.
  • User: کاربری که Job زمان‌بندی می‌کند و وضعیت Jobهای خودش را می‌بیند.

در مصاحبه واقعی، همین یک فهرست کوتاه کافی است. مهم این است که آن را با مصاحبه‌کننده مرور کنید.

طراحی API

هدف این است که نیازمندی‌های اصلی را یکی‌یکی مرور کنید و API لازم برای هر کدام را تعریف کنید. معمولاً رابطه‌ی یک‌به‌یک وجود دارد، ولی گاهی برای یک نیازمندی به چند Endpoint نیاز است.

اول، ایجاد Job:

1
2
3
4
5
6
7
8
9

POST /jobs
{
  "task_id": "send_email",
  "schedule": "0 10 * * *",
  "parameters": {
    "to": "[email protected]",
    "subject": "Daily Report"
  }
}

بعد، پرس‌وجوی وضعیت Jobها:

1

GET /jobs?user_id={user_id}&status={status}&start_time={start_time}&end_time={end_time} -> Job[]

جریان داده

پیش از ورود به طراحی فنی، بهتر است مسیر حرکت داده در سیستم را مرور کنیم: از لحظه‌ای که درخواست وارد می‌شود تا تولید خروجی نهایی. این کار سه فایده دارد. اول، پیش از پرداختن به جزئیات پیاده‌سازی با مصاحبه‌کننده درباره عملکرد اصلی هم‌نظر می‌شویم. دوم، نقشه‌راهی برای تصمیم‌های طراحی سطح بالا به دست می‌آید. سوم، گلوگاه‌های احتمالی را قبل از انتخاب معماری شناسایی می‌کنیم.

  1. کاربر با مشخص کردن Task، زمان اجرا و پارامترها یک Job زمان‌بندی می‌کند.
  2. Job در سیستم ذخیره می‌شود.
  3. یک Worker در زمان برنامه‌ریزی‌شده Job را برمی‌دارد و اجرا می‌کند.
  4. اگر اجرا شکست بخورد، با Exponential Backoff دوباره تلاش می‌شود.
  5. وضعیت Job در سیستم به‌روزرسانی می‌شود.

این جریان عمداً ساده است. هرچه جلوتر برویم کامل‌ترش می‌کنیم، ولی شروع ساده و تکمیل تدریجی روش درستی است.

طراحی سطح بالا

با ساخت یک MVP شروع می‌کنیم که نیازمندی‌های عملکردی اصلی را پوشش دهد. فعلاً لازم نیست مقیاس‌پذیر یا کامل باشد. این فقط پایه‌ای است که بعداً روی آن بنا می‌کنیم.

۱) زمان‌بندی Job برای اجرای فوری، آینده یا تکرارشونده

کاربر هنگام زمان‌بندی، Task، زمان‌بندی و پارامترها را مشخص می‌کند. مراحل کار این است:

  1. کاربر درخواستی به Endpoint ‏/jobs می‌فرستد که شامل شناسه Task (کدام کار اجرا شود)، Schedule (چه زمانی) و Parameters (ورودی‌های لازم) است.
  2. Job را با وضعیت PENDING در پایگاه داده ذخیره می‌کنیم.

این کار سه مزیت دارد: سابقه‌ی پایداری از همه Jobها داریم، در صورت خرابی سیستم می‌توانیم Jobها را بازیابی کنیم، و وضعیت Job را در تمام چرخه عمرش دنبال می‌کنیم.

درباره انتخاب پایگاه داده، حقیقت این است که هر پایگاه داده مدرنی جواب می‌دهد. چون به Strong Consistency نیاز نداریم و داده‌ها رابطه‌ی کمی با هم دارند، من یک Key-Value Store انعطاف‌پذیر مانند DynamoDB را انتخاب می‌کنم تا مقیاس‌دهی بعدی ساده‌تر شود. اگر شرکت راه‌حل‌های متن‌باز را ترجیح می‌دهد، Cassandra گزینه‌ی مناسبی است. Postgres یا MySQL هم به همان اندازه خوب‌اند، فقط بعداً باید بیشتر مراقب مقیاس‌دهی باشید.

با یک طرح ساده برای جدول Jobs شروع می‌کنیم تا ببینیم چرا نیاز به اصلاح دارد:

1
2
3
4
5
6
7
8
9
10
11

{
  "job_id": "123e4567-e89b-12d3-a456-426614174000",
  "user_id": "user_123",
  "task_id": "send_email",
  "scheduled_at": 1715548800,
  "parameters": {
    "to": "[email protected]",
    "subject": "Daily Report"
  },
  "status": "PENDING"
}

این ساختار برای Jobهای یک‌بارمصرف خوب است، اما با Jobهای تکرارشونده مشکل پیدا می‌کند. گزارش ایمیلی روزانه‌ای را در نظر بگیرید که هر روز ساعت ۱۰ صبح اجرا می‌شود. می‌توانیم عبارت CRON یعنی 0 10 * * * را در جدول ذخیره کنیم، اما چطور بفهمیم کدام Jobها در چند دقیقه‌ی آینده باید اجرا شوند؟ باید همه‌ی عبارت‌های CRON را در پایگاه داده ارزیابی کنیم، که واضح است مقیاس‌پذیر نیست.

نکته‌ی کلیدی اینجاست که باید تعریف Job را از نمونه‌های اجرای آن جدا کنیم. تقویم را در نظر بگیرید: رویدادی دارید که هر دوشنبه تکرار می‌شود، اما در اپلیکیشن تقویم، هر دوشنبه یک نمونه‌ی جداگانه از آن را می‌بینید. دقیقاً همین چیزی است که نیاز داریم.

داده‌ها را به دو جدول تقسیم می‌کنیم. جدول Jobs تعریف Job را نگه می‌دارد:

1
2
3
4
5
6
7
8
9
10
11
12
13

{
  "job_id": "123e4567-e89b-12d3-a456-426614174000",  // Partition key for easy lookup by job_id
  "user_id": "user_123",
  "task_id": "send_email",
  "schedule": {
    "type": "CRON" | "DATE",
    "expression": "0 10 * * *"  // Every day at 10:00 AM for CRON, specific date for DATE
  },
  "parameters": {
    "to": "[email protected]",
    "subject": "Daily Report"
  }
}

جدول Executions هر بار اجرای یک Job را ردیابی می‌کند:

1
2
3
4
5
6
7
8

{
  "time_bucket": 1715547600,  // Partition key (Unix timestamp rounded down to hour)
  "execution_time": "1715548800-123e4567-e89b-12d3-a456-426614174000",  // Sort key (exact execution time and the jobId since partition key and sort key must be unique)
  "job_id": "123e4567-e89b-12d3-a456-426614174000",
  "user_id": "user_123",
  "status": "PENDING",
  "attempt": 0
}

با استفاده از Time Bucket (زمان یونیکس گردشده به ساعت) به‌عنوان Partition Key، هم جستجوی کارآمد داریم و هم از Hot Partition دوری می‌کنیم. مثلاً برای پیدا کردن Jobهایی که در چند دقیقه‌ی آینده اجرا می‌شوند، فقط باید Bucket ساعت فعلی و احتمالاً ساعت بعد را بخوانیم. محاسبه‌ی Bucket ساده است:

1

time_bucket = (execution_time // 3600) * 3600  # Round down to nearest hour

نتیجه این است که برای یافتن Jobهای نزدیک فقط ۱ یا ۲ پارتیشن را می‌خوانیم، و Bucketهای ساعتی باعث می‌شوند Jobها خوب بین پارتیشن‌ها پخش شوند و نقطه‌ی داغ ایجاد نشود. همچنین وقتی یک Job تکرارشونده اجرا می‌شود، با محاسبه‌ی زمان اجرای بعدی و ساخت یک رکورد جدید در جدول Executions، اجرای بعدی را زمان‌بندی می‌کنیم. تعریف Job ثابت می‌ماند و فقط نمونه‌های اجرای جدید ساخته می‌شوند.

جدا کردن «تعریف» از «نمونه‌ها» در طراحی سیستم الگوی رایجی است. آن را در سیستم‌های تقویم (تعریف رویداد در برابر رخدادها)، سیستم‌های اعلان (قالب در برابر اعلان‌های منفرد) و جاهای دیگر می‌بینید. این روش خوبی برای مدیریت رفتارهای تکرارشونده یا الگومحور است.

وقتی یک Worker آماده‌ی اجرا می‌شود، از جدول Executions رکوردهایی را می‌خواند که:

  • execution_time آن‌ها در چند دقیقه‌ی آینده است.
  • وضعیتشان PENDING است.

سپس Worker جزئیات کامل Job را از جدول Jobs می‌خواند و آن را اجرا می‌کند.

این رویکرد ساده شروع خوبی است، اما اینجا متوقف نشوید. در Deep Dive آن را به‌طور جدی گسترش می‌دهیم تا سیستم Scalable و Fault Tolerant باشد و شکست‌های Job را درست مدیریت کند.

۲) پایش وضعیت Jobها

اول، نکته‌ی بدیهی: وقتی Job اجرا می‌شود، وضعیت آن را در جدول Executions به یکی از مقادیر COMPLETED، FAILED، IN_PROGRESS و مانند آن به‌روزرسانی می‌کنیم.

اما چطور وضعیت همه‌ی Jobهای یک کاربر را بخوانیم؟ در طراحی فعلی، جستجو بر اساس user_id فرایند دومرحله‌ای ناکارآمدی دارد:

  1. جدول Jobs را برای پیدا کردن همه‌ی job_idهای کاربر جستجو کنیم (که یعنی Full Table Scan).
  2. برای هر کدام، جدول Executions را برای دیدن وضعیت بخوانیم.

برای حل این مشکل، یک Global Secondary Index یا همان GSI روی جدول Executions می‌گذاریم:

  • Partition Key: ‏user_id
  • Sort Key: ‏execution_time به‌همراه job_id

این GSI اجازه می‌دهد همه‌ی اجراهای یک کاربر را کارآمد پیدا کنیم، بر اساس زمان اجرا مرتب کنیم، Pagination داشته باشیم و در صورت نیاز بر اساس وضعیت فیلتر کنیم.

GSI مقداری سربار و هزینه اضافه می‌کند، اما برای پشتیبانی از درخواست‌های کاربر بدون آسیب به الگوی دسترسی اصلی (پیدا کردن Jobهای نزدیک به اجرا) معامله‌ی ارزشمندی است. این الگوی رایجی در DynamoDB و پایگاه‌های NoSQL مشابه است: به‌جای Denormalize کردن داده‌ها، چند الگوی دسترسی را با GSI پشتیبانی می‌کنید. جایگزین آن کپی کردن user_id در جدول پایه است که مدل داده را پیچیده‌تر و نگهداری آن را سخت‌تر می‌کند.

حالا کاربر فقط باید GSI را با user_id پرس‌وجو کند و فهرستی از اجراها را مرتب‌شده بر اساس execution_time بگیرد.

Deep Dive های احتمالی

۱) چطور اجرای Job را تا ۲ ثانیه بعد از زمان برنامه‌ریزی‌شده تضمین کنیم؟

طراحی فعلی محدودیت‌هایی دارد که اجازه نمی‌دهد به دقت لازم برسیم.

مهم‌ترین محدودیت این است که هر چند دقیقه یک بار از پایگاه داده می‌پرسیم کدام Jobها نزدیک اجرا هستند. فاصله‌ی اجرای این Cron، سقف دقت ما را تعیین می‌کند. اگر Cron هر ۲ دقیقه اجرا شود، یک Job می‌تواند تا ۲ دقیقه زودتر یا دیرتر اجرا شود.

پس شاید فکر کنیم Cron را بیشتر اجرا کنیم. اما برای رسیدن به دقت ۲ ثانیه باید آن را هر ۲ ثانیه یا کمتر اجرا کنیم، که به چند دلیل جواب نمی‌دهد:

  • با ۱۰ هزار Job در ثانیه، هر Query باید حدود ۲۰ هزار Job را بخواند و پردازش کند (Jobهای ۲ ثانیه‌ی آینده). این حجم برای جابه‌جایی و پردازش مکرر زیاد است.
  • حتی با Index مناسب، Query برای ۲۰ هزار Job ممکن است چند صد میلی‌ثانیه طول بکشد. با Latency شبکه و زمان Serialization، ممکن است ۵۰۰ میلی‌ثانیه فقط صرف گرفتن داده‌ی Job شود.
  • بعد از بازیابی Jobها هم زمان لازم است تا آن‌ها را مقداردهی اولیه کنیم، بین Workerها پخش کنیم و اجرا را شروع کنیم. این سربار پنجره‌ی دقیق ما را باز هم کوچک‌تر می‌کند.
  • اجرای Queryهای بزرگ در هر ۲ ثانیه بار قابل‌توجهی روی پایگاه داده می‌گذارد و می‌تواند روی سایر عملیات و پایداری کل سیستم اثر بگذارد.

پس این روش خوبی نیست. در عوض می‌توانیم هوشمندانه‌تر عمل کنیم و معماری زمان‌بندی دولایه معرفی کنیم که Durability را با دقت ترکیب می‌کند.

فاز اول، Query از پایگاه داده: مانند طراحی فعلی، جدول Executions را برای Jobهایی که در ۵ دقیقه‌ی آینده اجرا می‌شوند می‌خوانیم (این ۵ دقیقه حاشیه‌ای برای Latency شبکه و مانند آن است).

فاز دوم، Message Queue: فهرست Jobهای برگردانده‌شده را به‌ترتیب زمان اجرا به یک Message Queue می‌فرستیم. Workerها به‌ترتیب Jobها را از صف برمی‌دارند و اجرا می‌کنند.

این رویکرد دولایه با جدا کردن Query پایگاه داده از اجرای Job مزیت‌های مهمی دارد. چون Query پایگاه داده فقط هر ۵ دقیقه یک بار اجرا می‌شود، بار پایگاه داده کم می‌شود، و دقت را در سطح Message Queue حفظ می‌کنیم. کارایی بالای صف و تضمین ترتیب باعث می‌شود Jobها را با دقت زیر یک ثانیه اجرا کنیم و سقف فرکانسی که در طراحی قبلی داشتیم برداشته شود.

خوب، پیشرفت کردیم. اما Jobهای جدیدی که زمان اجرایشان کمتر از ۵ دقیقه‌ی دیگر است چه می‌شوند؟ اگر فقط در پایگاه داده نوشته شوند، Cron پنج‌دقیقه‌ای آن‌ها را نمی‌بیند. یا کلاً از دست می‌روند، یا در اجرای بعدی Cron، که ممکن است تا ۵ دقیقه بعد باشد، برداشته می‌شوند.

می‌توانیم Job را مستقیم در Message Queue بگذاریم، اما این مشکل تازه‌ای می‌سازد. صف‌های مبتنی بر Log مانند Kafka پیام‌ها را به‌ترتیب ورود پردازش می‌کنند، پس هر Job جدید به انتهای صف می‌رود. یعنی اگر پنجره‌ی ۵ دقیقه‌ای از Jobها در صف باشد، Job جدید پشت همه‌ی آن‌ها منتظر می‌ماند، حتی اگر زمان اجرایش زودتر باشد. این کار رسیدن به دقت ۲ ثانیه را غیرممکن می‌کند.

به‌جای آن، صف مبتنی بر Log را با سیستم صفی عوض می‌کنیم که بتواند Jobها را بر اساس زمان اجرا اولویت‌بندی کند. گزینه‌های اصلی را بررسی می‌کنیم.

گزینه اول، Redis Sorted Set (ZSET): با استفاده از Timestamp اجرا به‌عنوان Score، یک Priority Queue ساده به دست می‌آید. هر Job را با زمان اجرایش به‌عنوان Score به مجموعه اضافه می‌کنیم و Workerها با گرفتن ورودی‌هایی که Score کمتر از Timestamp فعلی دارند، Jobهای نوبتی را پیدا می‌کنند.

چالش‌ها: Redis با Latency زیر میلی‌ثانیه و عملیات Atomic عملکرد عالی دارد، اما در محیط توزیع‌شده پیچیدگی عملیاتی قابل‌توجهی دارد. باید منطق Retry را خودتان پیاده کنید، سناریوهای شکست را مدیریت کنید و Redis را طوری اداره کنید که Single Point of Failure نداشته باشد. این راه‌حل در مقیاس متوسط خوب جواب می‌دهد، اما برای اجرای قابل‌اعتماد در Production به تخصص عملیاتی و کد سفارشی زیادی نیاز دارد.

گزینه دوم، RabbitMQ: یک Message Broker قوی است که با افزونه‌ی Delayed Message Exchange امکان زمان‌بندی دارد. Persistence پیام‌ها را به‌صورت پیش‌فرض مدیریت می‌کند و Publisher Confirm برای تحویل قابل‌اعتماد دارد. Jobها با انتشار پیام همراه پارامتر تأخیر زمان‌بندی می‌شوند و Clustering در RabbitMQ، High Availability فراهم می‌کند.

چالش‌ها: RabbitMQ بالغ است، اما قابلیت پیام تأخیری آن بومی نیست و با افزونه کار می‌کند، که در مقیاس بالاتر ممکن است کمتر قابل‌اعتماد باشد. منطق Retry را باید دستی پیاده کنید، سربار عملیاتی از سرویس‌های مدیریت‌شده بیشتر است، مقیاس‌دهی افقی به پیکربندی و پایش دقیق Cluster نیاز دارد و اجرای پیام‌های تأخیری ممکن است به اندازه‌ی سیستم‌های زمان‌بندی تخصصی دقیق نباشد.

گزینه سوم، Amazon SQS: سرویس صف کاملاً مدیریت‌شده با پشتیبانی بومی از تحویل تأخیری پیام (SQS Delivery Delay). هنگام زمان‌بندی Job فقط پیام را با مقدار تأخیر به SQS می‌فرستیم و بقیه‌ی کار با SQS است. مثلاً برای اجرای Job در ۱۰ ثانیه‌ی آینده، پیام را با ۱۰ ثانیه تأخیر می‌فرستیم و SQS پس از آن، پیام را به Worker تحویل می‌دهد.

این روش نیاز به مدیریت زیرساخت را حذف می‌کند و تقریباً همه‌ی ویژگی‌های لازم را دارد. Delivery Delay تضمین می‌کند پیام‌ها تا زمان اجرای برنامه‌ریزی‌شده نامرئی بمانند. Visibility Timeout شکست Workerها را بعد از دریافت پیام مدیریت می‌کند. Dead Letter Queue هم Jobهای ناموفق را برای بررسی نگه می‌دارد. High Availability در چند Availability Zone فراهم است و سرویس خودکار با بار مقیاس می‌گیرد. قابلیت بومی پیام تأخیری، SQS را عملاً به یک Priority Queue توزیع‌شده تبدیل می‌کند، بدون پیچیدگی عملیاتی نگهداری آن.

برای مورد ما SQS بهترین انتخاب است، به این دلایل: پشتیبانی بومی از تحویل تأخیری (که آن را عملاً به صف اولویت‌دار بر اساس زمان تبدیل می‌کند)، مدیریت خودکار شکست Worker با Visibility Timeout، و مقیاس‌پذیری عالی برای بار ۱۰ هزار Job در ثانیه.

معماری دولایه‌ی نهایی به این صورت کار می‌کند:

  1. کاربر یک Job جدید می‌سازد و در پایگاه داده نوشته می‌شود.
  2. یک Cron Job هر ۵ دقیقه اجرا می‌شود و از پایگاه داده، Jobهایی را که در ۵ دقیقه‌ی آینده اجرا می‌شوند می‌خواند.
  3. Cron Job این Jobها را با مقدار تأخیر مناسب به SQS می‌فرستد.
  4. Workerها وقتی آماده‌ی اجرا هستند، پیام‌ها را از SQS دریافت می‌کنند.
  5. اگر Job جدیدی با زمان اجرای کمتر از ۵ دقیقه‌ی بعد ساخته شود، مستقیماً با تأخیر مناسب به SQS فرستاده می‌شود.

به یاد داشته باشید که بعضی مصاحبه‌کننده‌ها یا شرکت‌ها استفاده از سرویس‌های مدیریت‌شده مثل SQS را نمی‌پسندند. در این صورت می‌توانید Priority Queue را با Redis یا یک Data Store مشابه پیاده کنید. فقط از مصاحبه‌کننده بپرسید چه چیزی مجاز و چه چیزی غیرمجاز است.

۲) چطور مقیاس‌پذیری برای ۱۰ هزار Job در ثانیه را تضمین کنیم؟

در هر مصاحبه، وقتی درباره مقیاس صحبت می‌کنید، پیشنهاد من این است که سیستم را از چپ به راست دنبال کنید، هر گلوگاه را پیدا کنید و یکی‌یکی رفعش کنید.

ایجاد Job

از ایجاد Job شروع می‌کنیم. اگر ایجاد Job به‌طور یکنواخت با اجرای Job هماهنگ باشد، یعنی در هر ثانیه ۱۰ هزار Job ساخته می‌شود. این بعید است چون Jobهای تکرارشونده هم داریم، ولی حد بالای مسئله است. احتمالاً در این مرحله از مصاحبه‌کننده می‌پرسم چند درصد Jobها تکرارشونده‌اند. اگر این نسبت کوچک باشد و بیشتر Jobها یک‌باره باشند، سرویس ایجاد Job می‌تواند گلوگاه شود.

برای نرخ بالای ایجاد Job می‌توانیم یک Message Queue مثل Kafka یا RabbitMQ بین API و سرویس ایجاد Job بگذاریم. این صف هنگام اوج ترافیک مثل Buffer عمل می‌کند و اجازه می‌دهد سرویس ایجاد Job با نرخ پایدار کار کند و زیر فشار نماند. همچنین با اضافه کردن نمونه‌های بیشتر سرویس به‌عنوان Consumer، مقیاس‌دهی افقی ممکن می‌شود و Durability هم تضمین می‌کند که در صورت بروز مشکل در سرویس، درخواست‌ها از بین نروند.

با این حال، اضافه کردن Message Queue بین API و سرویس ایجاد Job احتمالاً طراحی را بیش از حد پیچیده می‌کند. پایگاه داده باید بتواند Throughput نوشتن را مستقیماً تحمل کند و همیشه به سرویسی جلوی آن نیاز داریم که افقی مقیاس بگیرد. مگر اینکه منطق تجاری سنگینی در سرویس ایجاد Job باشد که بخواهیم آن را جدا کنیم، بهتر است معماری را ساده نگه داریم و خود سرویس را مقیاس بدهیم.

در مصاحبه ممکن است این الگو بعضی مصاحبه‌کننده‌ها را تحت تأثیر قرار دهد، اما راه‌حل‌های ساده‌تر اغلب بهترند. پیش از افزودن اجزای اضافی، روی مقیاس‌دهی درست پایگاه داده و لایه‌ی سرویس تمرکز کنید. من آن را در نمودار می‌گذارم ولی اختیاری در نظر می‌گیرم.

پایگاه داده Jobها

همان‌طور که گفتیم، DynamoDB یا Cassandra را برای جدول‌های Jobs و Executions انتخاب کردیم. این انتخاب برای مقیاس مناسب است، چون DynamoDB می‌تواند هزاران عملیات نوشتن در ثانیه را در هر پارتیشن تحمل کند. کلیدهای پارتیشن را هم از قبل بهینه انتخاب کرده‌ایم:

  • جدول Jobs: پارتیشن‌بندی بر اساس job_id برای توزیع بار بین Jobها.
  • جدول Executions: پارتیشن‌بندی بر اساس زمان اجرا (Time Bucket) برای Query کارآمد Jobهای آینده.

با Provisioning مناسب، این جدول‌ها به‌راحتی بار ۱۰ هزار عملیات در ثانیه را تحمل می‌کنند و برای افزایش تعداد Jobها تقریباً بی‌نهایت مقیاس می‌گیرند.

نکته‌ی دیگر اینکه بعد از اجرای Job باید رکورد آن را برای Query کاربران نگه داریم. اما پس از یک مدت منطقی (مثلاً ۱ سال) می‌توانیم آن را به ذخیره‌سازی ارزان‌تر مثل S3 منتقل کنیم.

ظرفیت Message Queue

یک محاسبه‌ی سریع انجام می‌دهیم تا نیازمندی‌های SQS را بفهمیم. برای هر پنجره‌ی ۵ دقیقه‌ای باید حدود ۳ میلیون Job پردازش کنیم (۱۰ هزار Job در ثانیه ضرب در ۳۰۰ ثانیه). اندازه‌ی پیام در SQS تا ۲۵۶ کیلوبایت می‌تواند باشد، ولی پیام‌های ما بسیار کوچک‌اند، حدود ۲۰۰ بایت، و فقط شناسه‌ی Job، زمان اجرا و Metadata دارند. یعنی در هر پنجره‌ی ۵ دقیقه‌ای فقط ۶۰۰ مگابایت داده.

SQS مقیاس‌دهی و توزیع پیام بین Consumerها را خودکار انجام می‌دهد که یکی از مزیت‌های بزرگ آن است. نگران Sharding یا Partitioning دستی نیستیم و AWS همه را زیر کاپوت مدیریت می‌کند. این سرویس می‌تواند نیاز ۱۰ هزار پیام در ثانیه را بدون پیکربندی خاصی تحمل کند، هرچند باید از AWS درخواست افزایش Quota بدهیم، چون سقف پیش‌فرض با Batching برابر ۳۰۰۰ پیام در ثانیه است.

ممکن است همچنان چند صف بخواهیم، اما فقط برای جداسازی کارکردی (مثلاً صف‌های جدا برای اولویت‌ها یا انواع مختلف Job)، نه برای مقیاس‌دهی.

اگر هم سراغ Priority Queue مبتنی بر Redis بروید، ۳ میلیون Job به‌راحتی در حافظه جا می‌شود و جای نگرانی نیست. در این حالت نگرانی اصلی شما Fault Tolerance در صورت از کار افتادن Redis خواهد بود.

Workerها

برای لایه‌ی Worker باید گزینه‌های محاسباتی را با دقت بررسی کنیم. دو انتخاب اصلی، Container (با ECS یا Kubernetes) و Lambda Function است.

  • Container: برای بارهای ثابت معمولاً مقرون‌به‌صرفه‌تر است و برای Jobهای طولانی مناسب‌تر است، چون بین اجراها State را نگه می‌دارد. در عوض سربار عملیاتی بیشتری دارد و مقیاس‌پذیری آن به اندازه‌ی گزینه‌های Serverless نیست.
  • Lambda Function: واقعاً Serverless است و هزینه‌ی عملیاتی کمی دارد. برای Jobهای کوتاه زیر ۱۵ دقیقه عالی است و می‌تواند فوراً و خودکار با حجم کار مقیاس بگیرد. عیب‌های اصلی آن Cold Start است که می‌تواند روی نیاز دقت ۲ ثانیه اثر بگذارد، و هزینه‌ی بیشتر برای بارهای ثابت و پرحجم مثل بار ما.

با توجه به نیازهای ما، یعنی پردازش ۱۰ هزار Job در ثانیه با دقت ۲ ثانیه روی بار ثابت و قابل‌پیش‌بینی، من Container با ECS و Auto Scaling را انتخاب می‌کنم، هرچند در اینجا جواب غلطی وجود ندارد. Containerها بهترین تعادل بین هزینه و سادگی عملیاتی را می‌دهند و نیازهای عملکردی ما را برآورده می‌کنند. این پیکربندی را می‌توان بهینه‌تر کرد:

  • استفاده از Spot Instance برای کاهش هزینه.
  • تنظیم Auto Scaling بر اساس عمق صف.
  • گرم نگه داشتن (Pre-warm) استخر Container برای بار پایه.
  • تنظیم سیاست‌های Scaling برای مدیریت جهش‌های غیرمنتظره‌ی تقاضا.

۳) چطور اجرای At-Least-Once را تضمین کنیم؟

نگرانی اصلی ما مدیریت شکست‌هاست. اگر Worker به هر دلیلی نتواند Job را پردازش کند، می‌خواهیم قبل از رها کردن آن چند بار دوباره تلاش شود (مثلاً حداکثر ۳ بار برای هر Job).

مهم است بدانیم Job روی Worker به یکی از دو دلیل شکست می‌خورد:

  • شکست قابل مشاهده: Job به‌صورت آشکار شکست می‌خورد. معمولاً به‌خاطر باگ در کد Task یا پارامترهای ورودی نادرست.
  • شکست نامرئی: Job بی‌صدا شکست می‌خورد. معمولاً یعنی خود Worker از کار افتاده است.

با حالت ساده‌تر شروع می‌کنیم.

شکست‌های قابل مشاهده

کد Task را در بلوک try/catch می‌گذاریم تا خطا را ثبت کنیم و Job را ناموفق علامت بزنیم. سپس می‌توانیم قبل از تسلیم شدن، چند بار با Exponential Backoff دوباره تلاش کنیم. هر بار شکست را در جدول Executions می‌نویسیم و وضعیت را به Retry به‌همراه تعداد تلاش‌های انجام‌شده به‌روزرسانی می‌کنیم.

بعد Job را به Message Queue (همان SQS) برمی‌گردانیم تا چند ثانیه بعد دوباره تلاش شود (تأخیر با هر شکست به‌صورت نمایی بیشتر می‌شود). اگر Job بعد از ۳ بار تلاش باز هم شکست بخورد، وضعیتش را در جدول Executions به FAILED تغییر می‌دهیم و دیگر تلاش مجدد نمی‌شود.

خوشبختانه SQS این الگو را با Exponential Backoff به‌صورت آماده پشتیبانی می‌کند و پیاده‌سازی را ساده می‌کند.

شکست‌های نامرئی

حالت جذاب‌تر این است. وقتی Worker Crash می‌کند یا پاسخ نمی‌دهد، به راهی قابل‌اعتماد برای تشخیص آن و تلاش مجدد برای Job نیاز داریم. چند رویکرد را بررسی می‌کنیم.

رویکرد اول، Health Check: هر Worker یک Endpoint سلامت مثل GET /health ارائه می‌دهد و یک سرویس پایش مرکزی هر چند ثانیه آن را بررسی می‌کند. اگر Worker به چند Health Check متوالی پاسخ ندهد، سرویس پایش آن را ناموفق علامت می‌زند و Jobهایش به Workerهای سالم داده می‌شود.

چالش‌ها: Health Check در سیستم توزیع‌شده با هزاران Worker خوب مقیاس نمی‌گیرد، چون سرویس پایش باید دائماً همه را Poll کند. مشکلات شبکه بین سرویس پایش و Workerها می‌تواند False Positive ایجاد کند و باعث Retry غیرضروری شود. این رویکرد زیرساخت اضافه‌ای برای پایش لازم دارد و برای جلوگیری از Race Condition، هماهنگی پیچیده‌ای بین Health Checker و Scheduler می‌خواهد. و اگر خود سرویس پایش از کار بیفتد، مشکل بزرگ‌تر می‌شود.

رویکرد دوم، Job Leasing: یک مکانیزم Distributed Lock با استفاده از پایگاه داده برای ردیابی مالکیت Job. وقتی Worker می‌خواهد Jobی را پردازش کند، ابتدا با به‌روزرسانی رکورد Job با شناسه‌ی خودش و یک Timestamp انقضا، یک Lease می‌گیرد.

مثلاً اگر Worker A بخواهد Job 123 را پردازش کند، پایگاه داده را به این شکل به‌روز می‌کند: «Job 123 تا ساعت 10:30:15 به Worker A اجاره داده شد». هنگام پردازش، Worker باید Lease را مرتب با به‌روزرسانی زمان انقضا تمدید کند. اگر Worker A در ساعت 10:30:00 هنوز مشغول باشد، Lease را تا 10:30:30 تمدید می‌کند و ۱۵ ثانیه‌ی دیگر فرصت می‌گیرد. اگر Worker A از کار بیفتد، تمدید نمی‌کند و پس از گذشتن زمان انقضا، Worker دیگری می‌تواند Lease را بگیرد و Job را دوباره امتحان کند.

چالش‌ها: این الگو برای مدیریت پیچیدگی‌های سیستم توزیع‌شده به پیاده‌سازی دقیق نیاز دارد. پایگاه داده باید عملیات تمدید مکرر Lease را پردازش کند، که در مقیاس ۱۰ هزار Job در ثانیه، با فرض بازه‌ی Lease برابر ۵ ثانیه، می‌تواند به ۵۰ هزار به‌روزرسانی Lease در ثانیه برسد.

هماهنگی ساعت بین Workerها هم مهم می‌شود. اگر ساعت Worker A روی 10:30:00 و ساعت Worker B روی 10:30:20 باشد، Worker B ممکن است Job را بدزدد در حالی که Worker A هنوز مشغول پردازش است. Network Partition هم پیچیدگی بیشتری اضافه می‌کند: اگر Worker A نتواند به پایگاه داده برای تمدید Lease برسد ولی همچنان Job را پردازش کند، ممکن است Worker دیگری همان Job را شروع کند و اجرای تکراری رخ دهد.

رویکرد سوم، SQS Visibility Timeout: SQS مکانیزم داخلی برای مدیریت شکست Worker دارد. وقتی Worker پیامی را از صف می‌گیرد، SQS آن پیام را برای مدت قابل‌تنظیمی از دید Workerهای دیگر پنهان می‌کند. Worker پیام را پردازش می‌کند و بعد از موفقیت آن را حذف می‌کند. اگر Worker از کار بیفتد یا در مدت Visibility Timeout پیام را پردازش نکند، SQS پیام را دوباره برای Workerهای دیگر قابل‌مشاهده می‌کند.

برای بازیابی سریع از شکست و در عین حال پشتیبانی از Jobهای طولانی‌تر، Visibility Timeout را کوتاه تنظیم می‌کنیم (مثلاً ۳۰ ثانیه) و Workerها به‌طور دوره‌ای با فراخوانی ChangeMessageVisibility یک Heartbeat می‌فرستند تا این مدت را تمدید کنند. مثلاً Workerی که Job پنج‌دقیقه‌ای را پردازش می‌کند، هر ۱۵ ثانیه Visibility Timeout را تمدید می‌کند. به این ترتیب، اگر Worker Crash کند، Workerهای دیگر ظرف ۳۰ ثانیه Job را برمی‌دارند و لازم نیست منتظر انقضای یک Timeout طولانی بمانند.

مزایا: این رویکرد شکست Worker را بدون زیرساخت اضافه یا هماهنگی پیچیده مدیریت می‌کند. ترکیب Visibility Timeout کوتاه و Heartbeat دوره‌ای، هم تشخیص سریع شکست را ممکن می‌کند و هم از Jobهای طولانی پشتیبانی می‌کند. پیام‌های ناموفق پس از تعداد قابل‌تنظیمی Retry خودکار به Dead Letter Queue می‌روند و مکانیزم Heartbeat کنترل دقیقی بر مالکیت Job می‌دهد، بدون اینکه پیچیدگی اضافه کند.

Idempotency و تلاش مجدد

در نهایت، یکی از پیامدهای اجرای At-Least-Once این است که کد Task باید Idempotent باشد. یعنی چند بار اجرای یک Job باید همان نتیجه‌ی یک بار اجرا را بدهد. چند راه برای مدیریت این موضوع وجود دارد.

رویکرد اول، بدون کنترل: هر بار که Job دریافت شد اجرایش کنیم. مثلاً اگر Job «ارسال ایمیل خوش‌آمدگویی» باشد، هر بار پردازش Job ایمیل دوباره ارسال می‌شود و ایمیل تکراری می‌رود. چالش: این می‌تواند مشکلات جدی در سازگاری داده ایجاد کند. Jobی که پول منتقل می‌کند ممکن است چند بار آن را منتقل کند و Jobی که شمارنده را به‌روز می‌کند ممکن است چند برابر اضافه کند. این برای اکثر برنامه‌های واقعی پذیرفتنی نیست.

رویکرد دوم، جدول Deduplication: قبل از اجرای Job، جدول Deduplication را بررسی می‌کنیم که آیا این اجرا قبلاً پردازش شده است یا نه. هر اجرای موفق را با شناسه‌ی یکتایی که از ترکیب job_id و Timestamp اجرا ساخته می‌شود ذخیره می‌کنیم. اگر رکورد مطابق پیدا شد، اجرا را رد می‌کنیم. چالش: به هر اجرای Job یک عملیات پایگاه داده اضافه می‌کند و نگهداری جدول دیگری لازم دارد. برای جلوگیری از رشد نامحدود باید جدول به‌صورت دوره‌ای پاکسازی شود. همچنین یک پنجره‌ی کوچک Race Condition بین بررسی و نوشتن در جدول وجود دارد.

رویکرد سوم، طراحی Idempotent: با استفاده از Idempotency Key و عملیات شرطی، Jobها را طوری طراحی می‌کنیم که ذاتاً Idempotent باشند. مثلاً به‌جای «شمارنده را یکی زیاد کن»، Job می‌تواند «مقدار شمارنده را روی X بگذار» باشد. به‌جای «ارسال ایمیل خوش‌آمدگویی»، ابتدا بررسی می‌کنیم که پرچم ارسال ایمیل خوش‌آمدگویی در پروفایل کاربر ثبت شده است یا نه. هر اجرای Job یک شناسه‌ی یکتا دارد که سرویس‌های پایین‌دستی می‌توانند برای حذف درخواست‌های تکراری استفاده کنند.

این قوی‌ترین رویکرد است، چون مسئولیت Idempotency را به خود Job می‌سپارد.

انتظار در هر سطح

Mid-level: از نامزدهای این سطح انتظار می‌رود یک طراحی سطح بالا با مدل داده‌ی واضح و منسجم ارائه دهند. بعد باید بتوانند به سؤال‌های مصاحبه‌کننده درباره Durability، زمان‌بندی دقیق و Retry پاسخ مناسب بدهند. انتظار ندارم بدون راهنمایی جواب درست را بدانند، ولی مهارت قوی حل مسئله و توانایی همکاری با مصاحبه‌کننده برای رسیدن به راه‌حل‌های منطقی را می‌خواهم ببینم.

Senior: از نامزدهای Senior خودمختاری بیشتری انتظار دارم. آن‌ها باید گفتگو را حول دو Deep Dive هدایت کنند و تجربه‌ی عملی با فناوری‌های مرتبط نشان دهند. به‌ویژه باید بتوانند نیاز به معماری دوفازی را توضیح دهند، حتی اگر Redis یا گزینه‌ی دیگری را برای Priority Queue انتخاب کنند. می‌خواهم ببینم مسائل را فعالانه شناسایی می‌کنند و سریع با سناریوهایی که پیش پایشان می‌گذارم سازگار می‌شوند، مثلاً «اگر Worker وسط اجرای Job از کار بیفتد چه می‌شود؟»

Staff و بالاتر: از نامزدهای Staff انتظار دارم راه‌اندازی و طراحی سطح بالا را سریع پشت سر بگذارند و بیشتر وقت را صرف Deep Dive کنند. باید بحث را فعالانه رهبری کنند، گلوگاه‌های اصلی سیستم را بشناسند و راه‌حل‌های مستدل پیشنهاد دهند. معمولاً درک عمیقی از دست‌کم یکی از فناوری‌های مطرح‌شده دارند و می‌توانند با تکیه بر تجربه‌ی عملی توضیح دهند چرا یک رویکرد از دیگری بهتر است. هرچند الزام سختی نیست، بهترین نامزدهای Staff می‌توانند چیزی، هر چقدر کوچک، به مصاحبه‌کننده یاد بدهند.

طراحی سیستم, مصاحبه
system design job scheduler SQS DynamoDB مصاحبه