Retry داخل Kafka Producer
retries=5، retry_backoff_ms=500، acks=all و enable_idempotence=True.
این مستند معماری، فلوی اجرا، و تصمیمات مهندسی زیرسیستم خزش Play Store را شرح می‌دهد. این زیرسیستم مسئول استخراج دوره‌ای (هر ۱ ساعت) آمار کلی و نقدهای هر اپلیکیشن فعال، و انتشار آن‌ها به Kafka برای مصرف توسط Storage Consumer است.
۱ تصمیم معماری کلیدی: چرا Threading و نه Asyncio؟
با اینکه این کار ذاتاً I/O-Bound است (بیشتر زمان صرف انتظار برای پاسخ شبکه می‌شود، نه پردازش)، که معمولاً محیط مناسبی برای asyncio است، این تصمیم عمداً گرفته نشد. دلیل: تمام وابستگی‌های این زیرسیستم — کتابخانه‌ی google-play-scraper، کتابخانه‌ی Kafka، و کتابخانه‌ی HTTP برای ارتباط با App Registry — به‌صورت Synchronous (Blocking) هستند. افزودن asyncio روی چنین پشته‌ای، فقط یک لایه‌ی مدیریتی اضافه می‌کند بدون فایده‌ی واقعی، در حالی که ThreadPoolExecutor مستقیماً و با پیچیدگی به‌مراتب کمتر همان کنترل همزمانی را فراهم می‌کند.
flowchart LR
A["Concurrency Model"] --> B{"وابستگی‌ها Sync هستند؟"}
B -->|بله| C["ThreadPoolExecutor
(انتخاب شده)"] B -->|خیر، Async-native| D["asyncio
(رد شده برای این پروژه)"] C --> E["پیچیدگی کم، مستقیم روی Blocking Library"] D --> F["نیاز به لایه‌ی ترجمه Sync↔Async
بدون فایده‌ی واقعی در این مقیاس"] style C fill:#16a34a,color:#fff style D fill:#dc2626,color:#fff
(انتخاب شده)"] B -->|خیر، Async-native| D["asyncio
(رد شده برای این پروژه)"] C --> E["پیچیدگی کم، مستقیم روی Blocking Library"] D --> F["نیاز به لایه‌ی ترجمه Sync↔Async
بدون فایده‌ی واقعی در این مقیاس"] style C fill:#16a34a,color:#fff style D fill:#dc2626,color:#fff
۲ تفکیک دو مفهوم: Concurrency Limit در برابر Rate Limit
این دو مفهوم عمداً با دو مکانیزم کاملاً مستقل پیاده‌سازی شده‌اند تا نقش هرکدام شفاف بماند.
flowchart TD
subgraph Concurrency["Concurrency Limit - چند کار همزمان در حال اجراست"]
TP["ThreadPoolExecutor(max_workers=5)"]
end
subgraph RateLimit["Rate Limit - چند درخواست واقعی در بازه زمانی"]
RL["RateLimiter (Token Bucket)
max_requests=10 / per_seconds=60
+ Jitter تصادفی"] end TP -->|هر Thread قبل از هر درخواست| RL RL --> PS["Google Play Store"] style TP fill:#2563eb,color:#fff style RL fill:#f59e0b,color:#fff
max_requests=10 / per_seconds=60
+ Jitter تصادفی"] end TP -->|هر Thread قبل از هر درخواست| RL RL --> PS["Google Play Store"] style TP fill:#2563eb,color:#fff style RL fill:#f59e0b,color:#fff
| مکانیزم | سوالی که پاسخ می‌دهد | پیاده‌سازی |
|---|---|---|
| Concurrency Limit | چند اپ همزمان در حال پردازش‌اند؟ | ThreadPoolExecutor(max_workers=5) |
| Rate Limit | در هر دقیقه چند درخواست واقعی به Play Store می‌رود؟ | RateLimiter با الگوی Token Bucket، Thread-safe (با Lock)، با Jitter تصادفی برای جلوگیری از الگوی درخواست کاملاً منظم |
| Retry | در صورت خطای موقتی شبکه چه باید کرد؟ | Exponential Backoff (2s → 4s → 8s)، محدود به خطاهای واقعاً موقتی (RequestException)؛ NotFoundError هرگز Retry نمی‌شود |
۳ معماری لایه‌ای و تزریق وابستگی (Dependency Injection)
برخلاف زیرسیستم App API (که بر پایه‌ی Active Record جنگو است)، این زیرسیستم از کلاس‌های پایتون خالص تشکیل شده که وابستگی‌هایشان از بیرون (Constructor) تزریق می‌شود. این تصمیم مستقیماً از استراتژی تست این زیرسیستم ناشی می‌شود: تمام تست‌ها با Mock کامل (بدون اتصال واقعی) نوشته شده‌اند، و DI دقیقاً همان مکانیزمی است که این جایگزینی را ممکن می‌سازد.
flowchart TD
Settings["Settings
(immutable, validated, از .env)"] --> Main["main.py
(Composition Root)"] Main --> RL[RateLimiter] Main --> PSC[PlayStoreClient] Main --> KP[CrawlerKafkaProducer] Main --> ARC[AppRegistryClient] RL --> PSC Main --> CS[CrawlerService] PSC --> CS KP --> CS ARC --> CS Main --> SCH[Scheduler] CS --> SCH style Settings fill:#7c3aed,color:#fff style Main fill:#2563eb,color:#fff style CS fill:#16a34a,color:#fff
(immutable, validated, از .env)"] --> Main["main.py
(Composition Root)"] Main --> RL[RateLimiter] Main --> PSC[PlayStoreClient] Main --> KP[CrawlerKafkaProducer] Main --> ARC[AppRegistryClient] RL --> PSC Main --> CS[CrawlerService] PSC --> CS KP --> CS ARC --> CS Main --> SCH[Scheduler] CS --> SCH style Settings fill:#7c3aed,color:#fff style Main fill:#2563eb,color:#fff style CS fill:#16a34a,color:#fff
Composition Root:
main.py تنها جایی در کل زیرسیستم است که تمام اشیاء واقعی ساخته و به‌هم متصل می‌شوند. تمام تنظیمات (Rate Limit، Retry، تعداد Worker، بازه‌ی زمانی کراول) از یک کلاس Settings واحد و Immutable خوانده می‌شوند که در ابتدای اجرا Validate می‌شود — یعنی مقدار نامعتبر محیطی (مثل عدد منفی) بلافاصله در لحظه‌ی شروع برنامه با خطای واضح متوقف می‌شود، نه ساعت‌ها بعد وسط یک دور کراول.۴ فلوی کامل یک دور کراول
sequenceDiagram
participant Scheduler
participant Service as CrawlerService
participant Registry as AppRegistryClient
participant Pool as ThreadPoolExecutor
participant PSClient as PlayStoreClient
participant RateLimiter
participant PlayStore as Google Play Store
participant Mapper as data_mapper
participant Kafka
Scheduler->>Service: run_crawl_cycle() [هر ۱ ساعت]
Service->>Registry: get_active_applications()
Registry-->>Service: [app1, app2, app3, ...]
alt لیست خالی
Service-->>Scheduler: لاگ اطلاعاتی، بدون خطا
else لیست غیرخالی
Service->>Pool: submit(crawl, app) برای هر اپ
par برای هر اپ (حداکثر ۵ همزمان)
Pool->>PSClient: get_app_details(package_name)
PSClient->>RateLimiter: acquire()
RateLimiter-->>PSClient: مجوز (احتمالاً پس از Sleep)
PSClient->>PlayStore: درخواست HTTP
alt خطای شبکه موقتی
PlayStore-->>PSClient: Timeout/ConnectionError
PSClient->>PSClient: Retry با Backoff (2s, 4s, 8s)
else اپ یافت نشد
PlayStore-->>PSClient: NotFoundError
PSClient-->>Pool: raise (بدون Retry)
else موفق
PlayStore-->>PSClient: raw data (camelCase)
PSClient->>Mapper: map_app_details(raw)
Mapper-->>PSClient: داده مطابق schema (snake_case + crawled_at)
PSClient->>Kafka: send_app_stats(package_name, mapped_data)
end
Pool->>PSClient: get_reviews(package_name, count=1000)
Note over PSClient: مسیر مستقل از app_details؛
شکست یکی مانع دیگری نمی‌شود PSClient->>Mapper: map_review(raw) برای هر ریویو PSClient->>Kafka: send_reviews(package_name, mapped_reviews) end Service->>Service: جمع‌آوری نتایج (as_completed) Service->>Kafka: flush() Service->>Scheduler: خلاصه لاگ (app_stats: X/N، reviews: Y/N) end
شکست یکی مانع دیگری نمی‌شود PSClient->>Mapper: map_review(raw) برای هر ریویو PSClient->>Kafka: send_reviews(package_name, mapped_reviews) end Service->>Service: جمع‌آوری نتایج (as_completed) Service->>Kafka: flush() Service->>Scheduler: خلاصه لاگ (app_stats: X/N، reviews: Y/N) end
۵ قرارداد داده: از کتابخانه تا پایگاه‌داده
داده‌ی خام کتابخانه (camelCase، بدون timestamp کراول) هرگز مستقیم به Kafka فرستاده نمی‌شود. یک لایه‌ی data_mapper مستقل، آن را به دقیقاً همان ساختار مورد انتظار جدول‌های پایگاه‌داده (طبق طراحی اسکیما) تبدیل می‌کند. این تصمیم Kafka را از فرمت داخلی یک کتابخانه‌ی خارجی مستقل نگه می‌دارد.
| خروجی خام کتابخانه | پیام نهایی در Kafka | دلیل تبدیل |
|---|---|---|
minInstalls, adSupported, ... | min_installs, ad_supported, ... | سازگاری با نام‌گذاری snake_case جدول‌های Postgres (camelCase بدون کوتیشن در Postgres به‌صورت خودکار lowercase و غیرقابل‌خواندن می‌شود) |
updated (Unix Timestamp عددی) | app_updated_at (رشته ISO 8601) | تطابق مستقیم با نوع DateTime ستون در جدول app_stats |
| — (وجود ندارد) | crawled_at (رشته ISO 8601، تولید‌شده در لحظه‌ی دریافت) | نیازمندی صریح سند پروژه: هر رکورد باید با timestamp لحظه‌ی خزش ثبت شود |
۶ استراتژی سه‌لایه‌ی Retry برای انتشار پیام در Kafka
نسخه‌ی اولیه‌ی Producer، پیام را فقط در بافر می‌گذاشت و تنها در پایان کل چرخه یک flush کلی انجام می‌داد — یعنی خطای تحویل یک پیام خاص، اغلب دیر یا حتی بی‌صدا دیده می‌شد. برای رفع این مشکل، انتشار پیام به سه لایه‌ی مستقل و مکمل تقسیم شد تا هیچ داده‌ای که با موفقیت از Play Store گرفته شده، به‌خاطر یک قطعی موقتی Kafka از دست نرود.
Retry سطح Application
max_attempts=3 با تأخیر 5s → 10s → 20s برای app-stats و residual reviewها.
Fail-Closed
پس از شکست retryها، خطا به سرویس برمی‌گردد؛ اپ فعلی رد می‌شود اما چرخه به اپ بعدی ادامه می‌دهد.
flowchart TD
A["send_app_stats / send_reviews فراخوانی می‌شود"] --> L1
subgraph L1["لایه ۱ - داخل خود Kafka Producer"]
direction TB
L1a["retries=5, retry_backoff_ms=500"]
L1b["acks=all: تایید تمام Replicaها"]
L1c["enable_idempotence=True: جلوگیری از پیام تکراری"]
end
L1 -->|"شکست پس از ۵ تلاش داخلی"| L2
subgraph L2["لایه ۲ - Retry سطح Application"]
direction TB
L2a["with_retry برای app-stats
(تکرار همان فراخوانی)"] L2b["run_residual_retry برای reviews
(فقط موارد failشده تکرار می‌شوند)"] L2c["max_attempts=3, delay=5s→10s→20s"] end L2 -->|"شکست پس از ۳ تلاش application"| L3["لایه ۳ - Fail-Closed
raise به crawler_service
لاگ + رد شدن از این اپ + ادامه به اپ بعدی"] L1 -->|"موفق"| OK["پیام تایید شد"] L2 -->|"موفق"| OK style L1 fill:#16a34a,color:#fff style L2 fill:#f59e0b,color:#fff style L3 fill:#dc2626,color:#fff style OK fill:#2563eb,color:#fff
(تکرار همان فراخوانی)"] L2b["run_residual_retry برای reviews
(فقط موارد failشده تکرار می‌شوند)"] L2c["max_attempts=3, delay=5s→10s→20s"] end L2 -->|"شکست پس از ۳ تلاش application"| L3["لایه ۳ - Fail-Closed
raise به crawler_service
لاگ + رد شدن از این اپ + ادامه به اپ بعدی"] L1 -->|"موفق"| OK["پیام تایید شد"] L2 -->|"موفق"| OK style L1 fill:#16a34a,color:#fff style L2 fill:#f59e0b,color:#fff style L3 fill:#dc2626,color:#fff style OK fill:#2563eb,color:#fff
| پارامتر | مقدار | چرا این مقدار |
|---|---|---|
kafka_producer_retries | 5 | لایه‌ی داخلی ارزان‌ترین جای Retry است (بدون تاخیر طولانی)؛ فرصت کافی به کتابخانه داده می‌شود تا مشکلات لحظه‌ای (چند میلی‌ثانیه‌ای) را خودش حل کند، پیش از ارجاع به لایه‌ی گران‌تر |
kafka_producer_retry_backoff_ms | 500 | مقدار پیش‌فرض استاندارد در اکوسیستم Kafka (سازگار با librdkafka و اکثر کلاینت‌های رسمی)؛ به‌اندازه‌ی کافی کوتاه که ۵ تلاش کل بیش از چند ثانیه طول نکشد |
kafka_producer_acks | all | تنها گزینه‌ای که Durability واقعی تضمین می‌کند (منتظر تایید تمام Replicaهای هماهنگ)؛ کندتر از acks=1 ولی چون فرکانس ارسال پایین است (چند ده پیام در ساعت، نه هزاران در ثانیه)، این هزینه در برابر تضمین از‌دست‌نرفتن داده کاملاً توجیه‌پذیر است |
kafka_producer_enable_idempotence | True | وقتی Retry داخلی فعال است، بدون این گزینه خطر واقعی ثبت پیام تکراری (Duplicate) در Kafka وجود دارد؛ فعال‌سازی آن عملاً هزینه‌ی منفی ندارد |
kafka_producer_request_timeout_ms | 5000 | تعادل بین تحمل نوسان طبیعی شبکه و جلوگیری از قفل ماندن طولانی یک Worker Thread از سقف محدود (۵ Thread) Concurrency |
kafka_producer_delivery_timeout_ms | 15000 | سقف زمانی کل مسیر یک پیام شامل تمام تلاش‌های داخلی؛ تضمین می‌کند حتی در بدترین حالت لایه‌ی ۱، پیام در یک زمان معقول (نه بی‌نهایت) به لایه‌ی ۲ ارجاع داده شود |
kafka_send_retry_max_attempts | 3 | رسیدن به این لایه یعنی کل بودجه‌ی لایه‌ی داخلی از قبل شکست خورده — نشانه‌ی یک مشکل جدی‌تر؛ به همین دلیل کمتر از ۵ تلاش لایه‌ی اول، هم‌راستا با الگوی PlayStoreClient |
kafka_send_retry_base_delay_seconds | 5.0 | بزرگ‌تر از تاخیر پایه‌ی Play Store (۲ ثانیه) چون مشکل در این نقطه جدی‌تر فرض می‌شود؛ صبر بیشتر پیش از تلاش مجدد، به‌جای فشار اضافه به سیستمی که احتمالاً از قبل تحت فشار است |
۷ فلوی دقیق ارسال ریویوها: Batch-Confirm با Residual Retry
برای ارسال ~۱۰۰۰ ریویوی هر اپ، یک راهکار دو-فاز طراحی شده تا هم بهینگی Batch حفظ شود و هم فقط پیام‌های واقعاً ناموفق دوباره ارسال شوند (نه کل دسته).
sequenceDiagram
participant CS as crawler_service
participant KP as CrawlerKafkaProducer
participant Producer as KafkaProducer (کتابخانه)
participant Kafka
CS->>KP: send_reviews(package_name, [r1..r1000])
Note over KP: فاز ۱ - Enqueue دسته‌ای (برای بهره از Batching)
loop برای هر review
KP->>Producer: send(topic, key, value)
Producer-->>KP: future
end
Note over KP: فاز ۲ - Await/Classify فقط futureهای همین Worker
loop برای هر future
KP->>Producer: future.get(timeout)
alt موفق
Producer-->>KP: ack
else شکست
Producer-->>KP: KafkaError
KP->>KP: افزودن review به لیست residual
end
end
alt residual خالی است
KP-->>CS: موفقیت کامل
else residual غیرخالی (مثلاً ۱۲ ریویو fail شد)
KP->>KP: run_residual_retry(فقط ۱۲ مورد residual)
Note over KP: تکرار فاز ۱+۲ فقط روی موارد failشده
(2s → 4s → 8s backoff) alt همه در نهایت موفق شدند KP-->>CS: موفقیت else residual پس از اتمام retry باقی ماند KP-->>CS: raise KafkaError (fail-closed) Note over CS: لاگ + رد شدن از reviews این اپ
app_stats همچنان مستقل و سالم است end end
(2s → 4s → 8s backoff) alt همه در نهایت موفق شدند KP-->>CS: موفقیت else residual پس از اتمام retry باقی ماند KP-->>CS: raise KafkaError (fail-closed) Note over CS: لاگ + رد شدن از reviews این اپ
app_stats همچنان مستقل و سالم است end end
نکته‌ی کلیدی: به‌جای
flush() مشترک روی کل Producer، هر Worker فقط روی futureهای مربوط به فراخوانی خودش get می‌زند — یعنی Attribution خطا همیشه دقیق و مختص همان اپ باقی می‌ماند.۸ استراتژی تست
flowchart TD
A["Unit Test (Mock)
اجرای پیش‌فرض، سریع، بدون شبکه"] --> B["Live Test
اجرای دستی، اتصال واقعی"] B --> C["End-to-End دستی
کل زنجیره با هم، قبل از تحویل"] style A fill:#16a34a,color:#fff style B fill:#f59e0b,color:#fff style C fill:#dc2626,color:#fff
اجرای پیش‌فرض، سریع، بدون شبکه"] --> B["Live Test
اجرای دستی، اتصال واقعی"] B --> C["End-to-End دستی
کل زنجیره با هم، قبل از تحویل"] style A fill:#16a34a,color:#fff style B fill:#f59e0b,color:#fff style C fill:#dc2626,color:#fff
| لایه | مثال | دلیل جداسازی |
|---|---|---|
| Unit Test (Mock) | آیا get_app_details فیلدها را درست فیلتر می‌کند؟ آیا Retry دقیقاً ۳ بار تلاش می‌کند؟ | سریع (کل مجموعه زیر ۲ ثانیه)، قطعی (بدون وابستگی به وضعیت اینترنت)، تکرارپذیر برای سناریوهای خطا |
| Live Test ( pytest -m live) | آیا اتصال واقعی به Google Play/Kafka همان ساختاری را برمی‌گرداند که در Mock فرض کرده‌ایم؟ | اعتبارسنجی فرض‌های Mock در برابر دنیای واقعی؛ اجرای دستی و کم‌دفعه برای جلوگیری از Rate Limit روی خودمان |
| End-to-End دستی | اجرای کامل main.py با زیرساخت واقعی و بررسی پیام‌های نهایی در Kafka | تنها راه تایید اینکه تمام اجزا (نه فقط هرکدام جدا) واقعاً با هم کار می‌کنند |
۹ تصمیمات کلیدی دیگر
یک پیام Kafka به‌ازای هر ریویو (نه یک پیام حاوی لیست کامل)
این تصمیم Storage Consumer آینده را ساده نگه می‌دارد — هر پیام مستقل قابل ingestion، Upsert بر اساس review_id، Replay، و اعتبارسنجی است، بدون نیاز به منطق باز کردن دسته (Batch-unpacking).
مدیریت مستقل خطای app_stats و reviews برای هر اپ
این دو مسیر در یک Thread مشترک (برای استفاده‌ی بهینه از همان Rate Limiter) اجرا می‌شوند، اما هرکدام try/except جداگانه دارند. شکست در گرفتن ریویوها هرگز مانع ارسال موفق آمار کلی همان اپ (و برعکس) نمی‌شود.
پیکربندی Immutable و Fail-Fast (Settings)
تمام مقادیر پیکربندی (Rate Limit، Retry، تعداد Worker، ...) در یک کلاس dataclass(frozen=True) واحد جمع شده‌اند که در __post_init__ اعتبارسنجی می‌شود. این تضمین می‌کند مقدار محیطی نامعتبر، بلافاصله در لحظه‌ی شروع (نه ساعت‌ها بعد) برنامه را متوقف کند.
جداسازی .env محیط توسعه از مقادیر Docker
فایل crawler/.env فقط برای اجرای مستقل (venv) روی هاست است (مثلاً KAFKA_BOOTSTRAP_SERVERS=localhost:9092)؛ در docker-compose.yml این مقادیر مستقیماً و جداگانه به آدرس‌های داخلی شبکه‌ی داکر (kafka:29092) تنظیم می‌شوند، چون این دو محیط اجرا نیازمند آدرس‌دهی متفاوتی هستند.
۱۰ ابزارها و تکنولوژی‌های استفاده‌شده
| ابزار | نقش |
|---|---|
| google-play-scraper | استخراج آمار کلی و نقدهای اپلیکیشن از Play Store |
| ThreadPoolExecutor (concurrent.futures) | اجرای همزمان کراول چند اپلیکیشن با سقف مشخص |
| kafka-python | انتشار پیام‌های آمار و نقد به Kafka |
| APScheduler (BlockingScheduler) | اجرای خودکار و دوره‌ای (هر ۱ ساعت) چرخه‌ی کراول |
| requests | ارتباط HTTP با App Registry API برای دریافت لیست اپلیکیشن‌های فعال |
| python-dotenv | بارگذاری پیکربندی محیطی در حالت توسعه (venv) |