Musa Yılmaz
·5 dk okuma

Django + Celery + Redis: Arka Plan Görevlerini Doğru Yapmak

djangoceleryredispython

Bir kullanıcı formu doldurup "Gönder"e bastığında ne olur? İdeal olarak sayfa anında yanıt verir. Ama arka planda bir e-posta gönderiliyorsa, SMTP sunucusu yavaş yanıt verdiğinde kullanıcı üç saniye ekrana bakar. SMTP sunucusu çökmüşse, kullanıcı hata sayfası görür — oysa formu aslında başarıyla kaydedilmiştir.

Çözüm, yavaş ve dış dünyaya bağımlı işleri istek döngüsünün dışına çıkarmaktır. Django ekosisteminde bunun standart yolu Celery'dir.

Neden Redis?

Celery bir görev kuyruğudur ve görevleri bir yerde saklaması gerekir. Bu yere "broker" denir. Redis ve RabbitMQ en yaygın seçeneklerdir.

Çoğu web projesi için Redis yeterlidir: kurulumu basittir, hafiftir ve muhtemelen zaten önbellek için kullanıyorsunuzdur. RabbitMQ daha güçlü garantiler sunar ama karmaşıklığı da beraberinde getirir — finansal işlem gibi hiçbir görevin kaybolmaması gereken senaryolarda tercih edilir.

Kurulum

pip install celery redis

Celery uygulamasını proje kökünde tanımlayın:

# config/celery.py
import os
from celery import Celery
 
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
 
app = Celery("config")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

autodiscover_tasks() her uygulamanın içindeki tasks.py dosyasını otomatik bulur — görevleri elle kaydetmek gerekmez.

Django başlarken Celery'nin de yüklenmesi için:

# config/__init__.py
from .celery import app as celery_app
 
__all__ = ("celery_app",)

Ayarlar:

# settings.py
CELERY_BROKER_URL = os.environ["REDIS_URL"]
CELERY_RESULT_BACKEND = os.environ["REDIS_URL"]
CELERY_TASK_SERIALIZER = "json"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TIMEZONE = "Europe/Istanbul"
 
# Görev 5 dakikadan uzun sürerse durdur
CELERY_TASK_TIME_LIMIT = 300

CELERY_TASK_SERIALIZER = "json" ayarını atlamayın. Celery'nin eski varsayılanı olan pickle, Python nesnelerini çalıştırılabilir şekilde serileştirir; broker'a erişim sağlayan biri bununla sunucuda kod çalıştırabilir. JSON bu riski taşımaz.

İlk görev

# notifications/tasks.py
from celery import shared_task
from django.core.mail import send_mail
 
@shared_task
def send_contact_notification(name: str, email: str, message: str) -> None:
    send_mail(
        subject=f"Yeni iletişim formu: {name}",
        message=message,
        from_email="noreply@example.com",
        recipient_list=["info@example.com"],
    )

Çağırmak:

# views.py
send_contact_notification.delay(name, email, message)

.delay() görevi kuyruğa atar ve anında döner. Kullanıcı beklemez.

Kritik kural: nesne değil, kimlik gönderin

Yeni başlayanların en sık yaptığı hata, göreve model nesnesi geçmektir:

# ❌ Yapmayın
process_order.delay(order)

Bunun iki sorunu var. Birincisi, nesnenin serileştirilmesi gerekir ve JSON serializer bunu yapamaz. İkincisi — daha sinsi olanı — görev çalıştığında nesnenin verisi eskimiş olabilir. Kuyrukta beklerken başka bir işlem kaydı değiştirmiş olabilir.

Doğrusu, kimliği geçip görevin içinde taze veriyi çekmektir:

# ✅ Doğru
process_order.delay(order.id)
 
@shared_task
def process_order(order_id: int) -> None:
    order = Order.objects.get(pk=order_id)
    # ...

Transaction ile ilgili tuzak

Bir görevi veritabanı işlemi (transaction) içinde tetiklerseniz, görev kayıt henüz commit edilmeden çalışmaya başlayabilir. Sonuç: Order.objects.get(pk=order_id) çağrısı DoesNotExist hatası verir.

Django'nun on_commit kancası bunu çözer:

from django.db import transaction
 
with transaction.atomic():
    order = Order.objects.create(...)
    transaction.on_commit(lambda: process_order.delay(order.id))

Artık görev, ancak işlem başarıyla tamamlandıktan sonra kuyruğa girer. Bu kalıbı görev tetiklediğiniz her yerde kullanmayı alışkanlık haline getirin.

Yeniden deneme (retry)

Dış servisler geçici olarak yanıt vermeyebilir. Görevin kendini yeniden denemesi çoğu zaman sorunu çözer:

@shared_task(
    bind=True,
    autoretry_for=(requests.RequestException,),
    retry_backoff=True,
    retry_kwargs={"max_retries": 5},
)
def notify_external_service(self, payload_id: int) -> None:
    payload = Payload.objects.get(pk=payload_id)
    response = requests.post(EXTERNAL_URL, json=payload.as_dict(), timeout=10)
    response.raise_for_status()

retry_backoff=True denemeler arasındaki süreyi katlanarak artırır (1s, 2s, 4s, 8s...). Bu önemlidir: sabit aralıkla denemek, zaten zorlanan bir servisi daha da yormaktır.

timeout=10 parametresini de atlamayın. Timeout'suz bir HTTP isteği süresiz asılı kalabilir ve worker'ınızı bloke eder.

Görevler idempotent olmalı

Bir görev birden fazla kez çalışabilir — retry sonrası, worker çökmesi sonrası, ya da mesaj iki kez teslim edildiğinde. Bu yüzden görevin iki kez çalışması zarar vermemeli.

Örneğin "kullanıcıya hoş geldin e-postası gönder" görevi iki kez çalışırsa kullanıcı iki e-posta alır. Basit bir bayrak bunu önler:

@shared_task
def send_welcome_email(user_id: int) -> None:
    user = User.objects.get(pk=user_id)
 
    if user.welcome_email_sent_at:
        return  # zaten gönderilmiş
 
    send_mail(...)
    User.objects.filter(pk=user_id).update(welcome_email_sent_at=timezone.now())

Zamanlanmış görevler

Celery Beat, cron benzeri periyodik görevler çalıştırır:

# settings.py
from celery.schedules import crontab
 
CELERY_BEAT_SCHEDULE = {
    "cleanup-expired-sessions": {
        "task": "accounts.tasks.cleanup_expired_sessions",
        "schedule": crontab(hour=3, minute=0),  # her gece 03:00
    },
}

Beat ayrı bir süreç olarak çalışır ve yalnızca bir kopyası çalışmalıdır. İki Beat süreci aynı anda çalışırsa her görev iki kez tetiklenir.

Docker ile çalıştırmak

Üç süreç gerekir: web, worker ve (periyodik görev varsa) beat.

services:
  redis:
    image: redis:7-alpine
 
  web:
    build: .
    command: gunicorn config.wsgi --bind 0.0.0.0:8000
    depends_on: [redis]
 
  worker:
    build: .
    command: celery -A config worker --loglevel=info --concurrency=2
    depends_on: [redis]
 
  beat:
    build: .
    command: celery -A config beat --loglevel=info
    depends_on: [redis]

--concurrency değerini sunucunuzun kaynaklarına göre ayarlayın. Her worker süreci bellekte Django uygulamasının bir kopyasını tutar; küçük bir sunucuda 8 worker açmak belleği tüketir.

Sunucuyu henüz kurmadıysanız AWS Lightsail'de Django deploy yazısı boş bir VM'de Docker'ı ayağa kaldırmayı anlatıyor. Next.js, Postgres ve Nginx'i Django'yla birlikte çalıştıran daha kapsamlı bir kurulum için Docker Compose ile Next.js + Django production kurulumu yazısına bakabilirsiniz.

İzleme

Görevlerin sessizce başarısız olması, olmamasından kötüdür. En azından loglara bakın:

docker compose logs -f worker

Daha kalıcı bir çözüm için Flower gibi bir izleme aracı kurulabilir, ancak panelin dışarıya açık kalmamasına dikkat edin — görev isimleri ve parametreleri iş mantığınız hakkında bilgi sızdırır.

Ne zaman gerek yok

Celery bir altyapı bileşenidir; Redis, worker süreci ve izleme ihtiyacı getirir. Şu durumlarda gerekmeyebilir:

  • Görev yeterince hızlıysa (100 ms altı) doğrudan çalıştırın
  • Tek bir zamanlanmış iş varsa, sistem cron'u yeterli olabilir
  • Sunucusuz bir platformdaysanız, platformun kendi kuyruk servisi daha uygun olabilir

Ama e-posta gönderimi, dış API çağrısı, dosya işleme veya rapor üretimi gibi işler söz konusuysa, bunları istek döngüsünden çıkarmak kullanıcı deneyimini doğrudan iyileştirir.