Advanced Celery Task Routing, Flower Monitoring and Error Handling

Advanced Celery Task Routing, Flower Monitoring and Error Handling

1. Konunun Mimari Önemi ve Temel Çalışma Prensibleri

Celery, Python için güçlü bir asynchronous task queue sistemidir. Bu, görevleri zaman uyumsuz olarak çalıştırmanıza ve dağıtılmış bir ortamda iş yükünü yönetmenize olanak tanır. Task routing, görevleri belirli worker’lar veya kuyruklara yönlendirerek sistem performansını ve iş yükünü optimize edebilir. Flower ise Celery iş yüklerini izlemek ve hata ayıklamak için güçlü bir monitöring aracıdır. Bu makalede, görev yönlendirme, monitörleme ve hata yönetimi konularında en gelişmiş teknikleri ve en iyi uygulamaları ele alacağız.

2. Kurumsal Projelerde Dikkat Edilmesi Gereken En İyi Pratikler (Best Practices)

• Görev Yönlendirme: Görevlerin doğru worker’lara yönlendirilmesi için dinamik routing politikaları kullanın. Örneğin, CPU yoğun görevler CPU yoğun worker’lara yönlendirilebilir.
• Monitörleme:

  Flower kullanarak kuyruk durumlarını, görev süresini ve hataları gerçek zamanlı olarak izleyin. Bu, sistem performansını ve sorunları hızlı bir şekilde tespit etmenize yardımcı olur.

• Hata Yönetimi: Görevlerde oluşabilecek hataları yakalamak ve yeniden denemeler için mekanizmalar oluşturun. Celery’de retry mekanizması bu amaca hizmet eder.
• Scalability: Sistem yükünü artırırken worker’ları uygun şekilde ölçeklendirin ve görevleri dağıtılmış bir ortamda eşit bir şekilde yükleyin.

# Celery configuration
app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

# Task routing
app.conf.task_routes = {
    'tasks.cpu_bound_task': {'queue': 'cpu_bound_queue'},
    'tasks.io_bound_task': {'queue': 'io_bound_queue'},
}

# Task example
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def cpu_bound_task(self, x, y):
    try:
        # CPU bound task logic
        result = x * y
        return result
    except Exception as exc:
        self.retry(exc=exc)

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def io_bound_task(self, url):
    try:
        # IO bound task logic
        response = requests.get(url)
        return response.text
    except Exception as exc:
        self.retry(exc=exc)