Merge branch 'feature/celery-integration': асинхронная обработка уведомлений через Celery

This commit is contained in:
Игорь Надеин 2026-01-08 16:14:06 +03:00
commit 56cfe3d332
14 changed files with 1218 additions and 18 deletions

View File

@ -0,0 +1,201 @@
# Celery Integration - Следующие шаги
## ✅ Что уже сделано
1. ✅ Добавлены зависимости в `requirements.txt`:
- `celery==5.5.3`
- `redis==5.2.1`
- `django-celery-beat==2.7.0`
- `django-celery-results==2.5.1`
- `flower==2.0.1`
2. ✅ Создан файл конфигурации `eusrr_backend/celery.py`
- Настроена автоматическая регистрация задач
- Добавлен debug task для тестирования
3. ✅ Обновлен `eusrr_backend/__init__.py`
- Автоматический импорт celery_app при запуске Django
4. ✅ Добавлены настройки Celery в `settings.py`:
- `CELERY_BROKER_URL` - Redis как брокер сообщений
- `CELERY_RESULT_BACKEND` - Redis для хранения результатов
- Настройка роутинга задач по очередям
- Таймауты и лимиты для workers
5. ✅ Добавлены приложения в `INSTALLED_APPS`:
- `django_celery_beat` - для периодических задач
- `django_celery_results` - для хранения результатов
6. ✅ Созданы Celery задачи в `notifications/tasks.py`:
- `send_notification_task` - отправка одного уведомления
- `send_bulk_notifications_task` - массовая отправка
- `cleanup_old_notifications_task` - очистка старых уведомлений
- `send_digest_email_task` - email-дайджесты
## 🔄 Следующие шаги (выполнить вручную)
### Шаг 1: Установка зависимостей
Запустите задачу в VS Code:
- **Pip: install requirements**
Или вручную:
```bash
cd backend
..\.venv\Scripts\pip install -r requirements.txt
```
### Шаг 2: Установка и запуск Redis
#### Вариант 1: Windows (рекомендуется Memurai)
1. Скачайте Memurai (Redis для Windows): https://www.memurai.com/
2. Установите и запустите как службу Windows
3. Redis будет доступен на `localhost:6379`
#### Вариант 2: Docker
```bash
docker run -d -p 6379:6379 --name redis redis:7-alpine
```
#### Проверка Redis:
```bash
redis-cli ping
# Должно вернуть: PONG
```
### Шаг 3: Применение миграций
Запустите:
```bash
cd backend
migrate_celery.bat
```
Или вручную:
```bash
..\.venv\Scripts\python manage.py migrate django_celery_beat
..\.venv\Scripts\python manage.py migrate django_celery_results
```
### Шаг 4: Тестирование Celery Worker
В отдельном терминале запустите worker:
```bash
cd backend
..\.venv\Scripts\celery -A eusrr_backend worker -l info --pool=solo
```
**Примечание**: На Windows используйте `--pool=solo` или `--pool=gevent`
### Шаг 5: Тестирование отправки задачи
В Django shell:
```python
..\.venv\Scripts\python manage.py shell
from celery import current_app
current_app.send_task('eusrr_backend.debug_task')
```
Вы должны увидеть выполнение задачи в логах worker.
### Шаг 6: Интеграция с NotificationService
Обновите `notifications/services.py`:
```python
from notifications.tasks import send_notification_task, send_bulk_notifications_task
class NotificationService:
@staticmethod
def create_notification(...):
# Вместо синхронного создания
send_notification_task.delay(
notification_type=notification_type,
user_id=user.id,
title=title,
message=message,
link=link,
sender_id=sender.id if sender else None,
metadata=metadata,
)
```
### Шаг 7: Запуск Flower (мониторинг)
```bash
cd backend
..\.venv\Scripts\celery -A eusrr_backend flower
```
Откройте http://localhost:5555 для просмотра статистики задач.
### Шаг 8: Настройка периодических задач
В Django Admin:
1. Перейдите в "Periodic Tasks"
2. Добавьте задачу для `notifications.cleanup_old_notifications`
3. Настройте интервал (например, раз в день)
Или программно в `eusrr_backend/celery.py`:
```python
from celery.schedules import crontab
app.conf.beat_schedule = {
'cleanup-old-notifications': {
'task': 'notifications.cleanup_old_notifications',
'schedule': crontab(hour=3, minute=0), # Каждый день в 3:00
'args': (90,) # Удалять уведомления старше 90 дней
},
}
```
Запустите beat scheduler:
```bash
..\.venv\Scripts\celery -A eusrr_backend beat -l info
```
## 📊 Ожидаемый результат
**До внедрения Celery:**
- Отправка сообщения в группу с 8 участниками: ~500ms
- Блокировка API response до отправки всех уведомлений
**После внедрения Celery:**
- Отправка сообщения: <50ms (только создание задачи)
- Уведомления обрабатываются асинхронно
- API мгновенно отвечает пользователю
## 🔍 Отладка
### Проверка статуса Celery
```bash
..\.venv\Scripts\celery -A eusrr_backend inspect active
..\.venv\Scripts\celery -A eusrr_backend inspect stats
```
### Проверка задач в очереди
```bash
..\.venv\Scripts\celery -A eusrr_backend inspect reserved
```
### Логи
- Worker логи: консоль где запущен worker
- Flower: http://localhost:5555
- Django логи: стандартные логи приложения
## 📚 Документация
- Подробная документация: `docs/guides/CELERY_SETUP.md`
- План внедрения (5 дней): `docs/in-progress/CELERY_INTEGRATION_PLAN.md`
- Celery docs: https://docs.celeryq.dev/
## ⚠️ Production Deployment
Для production:
1. Используйте Redis Cluster или Redis Sentinel
2. Запускайте несколько workers с supervisord/systemd
3. Настройте мониторинг (Flower + Prometheus)
4. Добавьте retry policies и dead letter queue
5. Настройте автоматический перезапуск workers
См. `docs/in-progress/CELERY_INTEGRATION_PLAN.md` для деталей.

View File

@ -78,7 +78,7 @@ def notify_event_cancelled(sender, instance, **kwargs):
# Создаем уведомления об отмене
for recipient in recipients:
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=recipient,
notification_type_code='event_cancelled',
title='Событие отменено',
@ -117,7 +117,7 @@ def notify_event_created(event):
)
for recipient in recipients:
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=recipient,
notification_type_code='event_created',
title='Новое событие в календаре',
@ -166,7 +166,7 @@ def notify_event_changed(event, changed_fields):
])
for recipient in recipients:
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=recipient,
notification_type_code='event_changed',
title='Событие изменено',

View File

@ -111,8 +111,8 @@ def create_message_notifications(sender, instance, created, **kwargs):
if user in participants_list and user.id != author.id:
# Проверяем настройки уведомлений (из предзагруженного словаря)
if notification_settings.get(user.id, True):
# Создаем уведомление об упоминании
NotificationService.create_notification(
# Создаем уведомление об упоминании (асинхронно)
NotificationService.create_notification_async(
recipient=user,
notification_type_code='chat_mention',
title=f'Вас упомянул {author.get_full_name() or author.username}',
@ -138,7 +138,7 @@ def create_message_notifications(sender, instance, created, **kwargs):
if original_author.id not in mentioned_user_ids:
# Проверяем настройки уведомлений (из предзагруженного словаря)
if notification_settings.get(original_author.id, True):
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=original_author,
notification_type_code='chat_reply',
title=f'{author.get_full_name() or author.username} ответил на ваше сообщение',
@ -197,7 +197,7 @@ def create_message_notifications(sender, instance, created, **kwargs):
if is_announcement:
metadata['is_announcement'] = True
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=recipient,
notification_type_code=notification_type,
title=title,
@ -238,7 +238,7 @@ def create_chat_added_notifications(sender, instance, action, pk_set, **kwargs):
if not notification_settings.get(user.id, True):
continue
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=user,
notification_type_code='chat_added_to_chat',
title=f'Вас добавили в чат',

View File

@ -333,7 +333,7 @@ def check_all_acknowledged(sender, instance, created, **kwargs):
total_recipients > 0 and
document.uploaded_by):
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=document.uploaded_by,
notification_type_code='document_signed_all',
title='Все ознакомились с документом',
@ -452,7 +452,7 @@ def create_document_ready_notification(document, recipient, send_immediately=Tru
else 'Администратор'
)
notification = NotificationService.create_notification(
notification = NotificationService.create_notification_async(
recipient=recipient,
notification_type_code='document_ready',
title='Новый документ на ознакомление',

View File

@ -0,0 +1,7 @@
"""
Инициализация проекта EUSRR
"""
# Это гарантирует, что приложение Celery загружается при старте Django
from .celery import app as celery_app
__all__ = ('celery_app',)

View File

@ -0,0 +1,24 @@
"""
Конфигурация Celery для проекта EUSRR
"""
import os
from celery import Celery
# Устанавливаем настройки Django по умолчанию
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'eusrr_backend.settings')
app = Celery('eusrr_backend')
# Загружаем конфигурацию из Django settings
# namespace='CELERY' означает что все настройки Celery в settings.py
# должны начинаться с префикса CELERY_
app.config_from_object('django.conf:settings', namespace='CELERY')
# Автоматически находит tasks.py в каждом приложении
app.autodiscover_tasks()
@app.task(bind=True, ignore_result=True)
def debug_task(self):
"""Тестовая задача для проверки работоспособности Celery"""
print(f'Request: {self.request!r}')

View File

@ -35,6 +35,10 @@ INSTALLED_APPS = [
"widget_tweaks",
"simple_history",
"rest_framework",
# Celery приложения
"django_celery_beat", # Периодические задачи
"django_celery_results", # Хранение результатов
# Основные приложения
"employees.apps.EmployeesConfig",
"api.apps.ApiConfig",
"hikcentral.apps.HikcentralConfig",
@ -460,4 +464,39 @@ VAPID_PRIVATE_KEY = os.getenv(
"VAPID_PRIVATE_KEY",
"MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgCNnpupg3xbtFUiOSUZ6L7s6puxuEjzR73kTL7v8bMvKhRANCAATE4rWcva-MoGCQYBnWmbgpG-68Ee4hyyb9A5DaWiTixEqLTjRE1rWQEMZyVVP2KPb_omKPu3VhHhdFRIzHBjJE"
)
VAPID_ADMIN_EMAIL = os.getenv("VAPID_ADMIN_EMAIL", "robotail-info@yandex.ru")
VAPID_ADMIN_EMAIL = os.getenv("VAPID_ADMIN_EMAIL", "robotail-info@yandex.ru")
# -----------------------------------------------------------------------------
# CELERY CONFIGURATION
# -----------------------------------------------------------------------------
CELERY_BROKER_URL = os.getenv('CELERY_BROKER_URL', 'redis://localhost:6379/0')
CELERY_RESULT_BACKEND = os.getenv('CELERY_RESULT_BACKEND', 'redis://localhost:6379/1')
# Сериализация
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
# Таймзона
CELERY_TIMEZONE = 'Europe/Moscow'
CELERY_ENABLE_UTC = True
# Мониторинг и отладка
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60 # 30 минут максимум на задачу
CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60 # Мягкий лимит - 25 минут
# Приоритеты очередей
CELERY_TASK_ROUTES = {
'notifications.tasks.*': {'queue': 'notifications'},
'documents.tasks.*': {'queue': 'default'},
'employees.tasks.*': {'queue': 'default'},
}
# Настройки для периодических задач (Celery Beat)
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
# Опции по умолчанию для всех задач
CELERY_TASK_ACKS_LATE = True # Подтверждаем выполнение после завершения
CELERY_WORKER_PREFETCH_MULTIPLIER = 1 # Берем по 1 задаче за раз
CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000 # Перезапуск worker после 1000 задач

View File

@ -49,7 +49,7 @@ def create_comment_notification(sender, instance, created, **kwargs):
# Уведомляем автора публикации (если комментарий не от него)
if post.author.id != comment_author.id:
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=post.author,
notification_type_code='feed_post_comment',
title='Новый комментарий к вашей публикации',
@ -90,7 +90,7 @@ def notify_new_post(post):
)
for recipient in recipients:
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=recipient,
notification_type_code='feed_new_post',
title=f'Новая публикация {context}',
@ -163,7 +163,7 @@ def notify_post_reaction(post, user):
"""
# Уведомляем автора (если лайк не от него)
if post.author.id != user.id:
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=post.author,
notification_type_code='feed_post_reaction',
title='Ваша публикация понравилась',

View File

@ -21,9 +21,104 @@ logger = logging.getLogger(__name__)
User = get_user_model()
def is_celery_available():
"""Проверка доступности Celery worker"""
try:
from celery import current_app
inspect = current_app.control.inspect()
active_workers = inspect.active()
return active_workers is not None and len(active_workers) > 0
except Exception:
return False
class NotificationService:
"""Сервис для создания и отправки уведомлений"""
@staticmethod
def create_notification_async(
recipient: User,
notification_type_code: str,
title: str,
message: str,
content_object=None,
action_url: str = '',
action_text: str = 'Посмотреть',
metadata: Optional[Dict[str, Any]] = None,
) -> bool:
"""
Создать уведомление асинхронно через Celery (рекомендуется)
Args:
recipient: Получатель уведомления
notification_type_code: Код типа уведомления
title: Заголовок
message: Текст сообщения
content_object: Связанный объект (опционально)
action_url: URL для действия
action_text: Текст кнопки действия
metadata: Дополнительные данные
Returns:
True если задача поставлена в очередь, False если fallback на sync
"""
try:
# Проверяем доступность Celery
if not is_celery_available():
logger.warning(
f"[create_notification_async] Celery недоступен, fallback на синхронный режим"
)
NotificationService.create_notification(
recipient=recipient,
notification_type_code=notification_type_code,
title=title,
message=message,
content_object=content_object,
action_url=action_url,
action_text=action_text,
metadata=metadata,
send_immediately=True,
)
return False
# Импортируем задачу
from notifications.tasks import send_notification_task
# Отправляем задачу в Celery
send_notification_task.delay(
notification_type=notification_type_code,
user_id=recipient.id,
title=title,
message=message,
link=action_url,
sender_id=None,
metadata=metadata or {},
)
logger.debug(
f"[create_notification_async] ✅ Задача отправлена в Celery: "
f"user={recipient.id}, type={notification_type_code}"
)
return True
except Exception as e:
logger.exception(
f"[create_notification_async] ❌ Ошибка при отправке в Celery: {e}"
)
# Fallback на синхронный режим
NotificationService.create_notification(
recipient=recipient,
notification_type_code=notification_type_code,
title=title,
message=message,
content_object=content_object,
action_url=action_url,
action_text=action_text,
metadata=metadata,
send_immediately=True,
)
return False
@staticmethod
def create_notification(
recipient: User,

View File

@ -0,0 +1,259 @@
"""
Celery задачи для асинхронной обработки уведомлений.
Эти задачи выполняются в фоновом режиме через Celery workers,
что позволяет значительно ускорить отклик API при отправке уведомлений.
"""
from celery import shared_task
from django.contrib.auth import get_user_model
from django.utils import timezone
from typing import List, Optional
import logging
logger = logging.getLogger(__name__)
User = get_user_model()
@shared_task(
name="notifications.send_notification",
bind=True,
max_retries=3,
default_retry_delay=60,
)
def send_notification_task(
self,
notification_type: str,
user_id: int,
title: str,
message: str,
link: Optional[str] = None,
sender_id: Optional[int] = None,
metadata: Optional[dict] = None,
):
"""
Асинхронная отправка уведомления одному пользователю.
Args:
notification_type: Тип уведомления (message, document, request и т.д.)
user_id: ID пользователя-получателя
title: Заголовок уведомления
message: Текст уведомления
link: Ссылка (опционально)
sender_id: ID отправителя (опционально)
metadata: Дополнительные данные (опционально)
"""
try:
from notifications.services import NotificationService
user = User.objects.get(id=user_id)
sender = User.objects.get(id=sender_id) if sender_id else None
# Вызываем синхронный метод создания уведомления
NotificationService.create_notification(
recipient=user,
notification_type_code=notification_type,
title=title,
message=message,
action_url=link or '',
metadata=metadata or {},
)
logger.debug(
f"Notification sent to user {user_id} (type: {notification_type})"
)
return {"status": "success", "user_id": user_id}
except User.DoesNotExist:
logger.error(f"User {user_id} not found for notification")
return {"status": "error", "reason": "user_not_found"}
except Exception as exc:
logger.exception(f"Failed to send notification to user {user_id}: {exc}")
# Повторяем попытку через 60 секунд
raise self.retry(exc=exc)
@shared_task(
name="notifications.send_bulk_notifications",
bind=True,
max_retries=3,
default_retry_delay=60,
)
def send_bulk_notifications_task(
self,
notification_type: str,
user_ids: List[int],
title: str,
message: str,
link: Optional[str] = None,
sender_id: Optional[int] = None,
metadata: Optional[dict] = None,
):
"""
Асинхронная массовая отправка уведомлений нескольким пользователям.
Оптимизирована для отправки большого количества уведомлений.
Args:
notification_type: Тип уведомления
user_ids: Список ID пользователей-получателей
title: Заголовок уведомления
message: Текст уведомления
link: Ссылка (опционально)
sender_id: ID отправителя (опционально)
metadata: Дополнительные данные (опционально)
"""
try:
from notifications.services import NotificationService
from notifications.models import Notification
# Получаем всех пользователей одним запросом
users = User.objects.filter(id__in=user_ids)
sender = User.objects.get(id=sender_id) if sender_id else None
# Создаем уведомления массово
notifications = []
for user in users:
notification = Notification(
user=user,
notification_type=notification_type,
title=title,
message=message,
link=link or "",
sender=sender,
metadata=metadata or {},
created_at=timezone.now(),
)
notifications.append(notification)
# Bulk create для производительности
Notification.objects.bulk_create(notifications)
# Отправляем WebSocket уведомления
for user in users:
NotificationService.send_websocket_notification(user.id)
logger.debug(
f"Bulk notification sent to {len(user_ids)} users (type: {notification_type})"
)
return {
"status": "success",
"sent_count": len(notifications),
"user_ids": user_ids,
}
except Exception as exc:
logger.exception(f"Failed to send bulk notifications: {exc}")
raise self.retry(exc=exc)
@shared_task(
name="notifications.cleanup_old_notifications",
bind=True,
)
def cleanup_old_notifications_task(self, days: int = 90):
"""
Периодическая задача для удаления старых прочитанных уведомлений.
Args:
days: Количество дней, после которого удаляются прочитанные уведомления
"""
try:
from notifications.models import Notification
from datetime import timedelta
cutoff_date = timezone.now() - timedelta(days=days)
deleted_count, _ = Notification.objects.filter(
is_read=True,
created_at__lt=cutoff_date
).delete()
logger.info(
f"Cleaned up {deleted_count} old notifications (older than {days} days)"
)
return {"status": "success", "deleted_count": deleted_count}
except Exception as exc:
logger.exception(f"Failed to cleanup old notifications: {exc}")
raise self.retry(exc=exc)
@shared_task(
name="notifications.send_digest_email",
bind=True,
)
def send_digest_email_task(self, user_id: int):
"""
Отправка email-дайджеста непрочитанных уведомлений пользователю.
Может вызываться периодически для пользователей с настроенными email-дайджестами.
Args:
user_id: ID пользователя
"""
try:
from notifications.models import Notification
from common.emails import send_email
user = User.objects.get(id=user_id)
# Получаем непрочитанные уведомления за последние 24 часа
cutoff_time = timezone.now() - timezone.timedelta(hours=24)
unread_notifications = Notification.objects.filter(
user=user,
is_read=False,
created_at__gte=cutoff_time
).order_by('-created_at')[:20]
if not unread_notifications:
logger.debug(f"No unread notifications for user {user_id}, skipping digest")
return {"status": "skipped", "reason": "no_notifications"}
# Формируем HTML для email
notification_html = ""
for notif in unread_notifications:
notification_html += f"""
<div style="margin-bottom: 15px; padding: 10px; border-left: 3px solid #007bff;">
<strong>{notif.title}</strong><br>
<span style="color: #666;">{notif.message}</span><br>
<small style="color: #999;">{notif.created_at.strftime('%H:%M, %d.%m.%Y')}</small>
</div>
"""
html_content = f"""
<html>
<body style="font-family: Arial, sans-serif; padding: 20px;">
<h2>У вас {len(unread_notifications)} непрочитанных уведомлений</h2>
<div style="margin-top: 20px;">
{notification_html}
</div>
<p style="margin-top: 30px;">
<a href="{settings.BASE_URL}/notifications/" style="color: #007bff;">
Посмотреть все уведомления
</a>
</p>
</body>
</html>
"""
send_email(
subject=f"У вас {len(unread_notifications)} новых уведомлений",
to_email=user.email,
html_content=html_content,
)
logger.info(f"Digest email sent to user {user_id} with {len(unread_notifications)} notifications")
return {
"status": "success",
"user_id": user_id,
"notification_count": len(unread_notifications),
}
except User.DoesNotExist:
logger.error(f"User {user_id} not found for digest email")
return {"status": "error", "reason": "user_not_found"}
except Exception as exc:
logger.exception(f"Failed to send digest email to user {user_id}: {exc}")
raise self.retry(exc=exc)

View File

@ -185,7 +185,7 @@ def create_comment_notification(sender, instance, created, **kwargs):
)
for recipient in recipients_set:
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=recipient,
notification_type_code="request_comment",
title=f"💬 Новый комментарий к заявлению от {employee_name}",
@ -373,10 +373,10 @@ def notify_new_request(request_obj):
notification_type = "request_new" # Используем общий тип
logger.info(
f"[notify_new_request] ➡️ Вызов NotificationService.create_notification для {recipient.username}"
f"[notify_new_request] ➡️ Вызов NotificationService.create_notification_async для {recipient.username}"
)
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=recipient,
notification_type_code=notification_type,
title=title,
@ -481,7 +481,7 @@ def notify_status_change(request_obj, old_status, new_status):
f"изменен: {old_status} → {new_status}"
)
NotificationService.create_notification(
NotificationService.create_notification_async(
recipient=recipient,
notification_type_code=notification_type,
title=title,

View File

@ -109,3 +109,8 @@ vine==5.1.0
wcwidth==0.2.14
yarl==1.22.0
zope.interface==8.1.1
# Celery и связанные пакеты (redis уже есть выше)
django-celery-beat>=2.7.0
django-celery-results>=2.5.1
flower>=2.0.1

290
docs/guides/CELERY_SETUP.md Normal file
View File

@ -0,0 +1,290 @@
# Инструкция по запуску Celery (Development)
## 📋 Предварительные требования
### 1. Установка Redis
**Windows:**
```bash
# Через WSL2
wsl --install
wsl
sudo apt update
sudo apt install redis-server
redis-server --daemonize yes
# Или скачать с https://github.com/microsoftarchive/redis/releases
```
**Linux/Mac:**
```bash
# Ubuntu/Debian
sudo apt install redis-server
sudo systemctl start redis
# MacOS
brew install redis
brew services start redis
```
**Docker (рекомендуется):**
```bash
docker run -d -p 6379:6379 --name redis redis:7-alpine
```
### 2. Проверка Redis
```bash
redis-cli ping
# Должно вернуть: PONG
```
---
## 🚀 Установка зависимостей
```bash
# Активировать виртуальное окружение
.venv/Scripts/activate # Windows
source .venv/bin/activate # Linux/Mac
# Установить новые пакеты
cd backend
pip install -r requirements.txt
```
Новые пакеты:
- `redis` - клиент для Redis
- `django-celery-beat` - периодические задачи
- `django-celery-results` - хранение результатов
- `flower` - веб-интерфейс мониторинга
---
## 🔧 Настройка окружения
### 1. Добавить в `.env`:
```env
# Redis & Celery
REDIS_HOST=localhost
REDIS_PORT=6379
CELERY_BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/1
```
### 2. Применить миграции (для django-celery-beat):
```bash
python manage.py migrate django_celery_beat
python manage.py migrate django_celery_results
```
---
## 🏃 Запуск Celery Worker
### Вариант 1: Базовый запуск
```bash
cd backend
celery -A eusrr_backend worker -l info
```
### Вариант 2: С указанием очередей
```bash
celery -A eusrr_backend worker -l info -Q default,notifications,emails
```
### Вариант 3: С autoscale (для production)
```bash
celery -A eusrr_backend worker -l info --autoscale=10,3
# 10 = максимум workers, 3 = минимум workers
```
---
## 🕐 Запуск Celery Beat (периодические задачи)
**Отдельное окно терминала:**
```bash
cd backend
celery -A eusrr_backend beat -l info
```
---
## 🌺 Запуск Flower (мониторинг)
**Отдельное окно терминала:**
```bash
cd backend
celery -A eusrr_backend flower --port=5555
```
Откройте в браузере: http://localhost:5555
---
## 🧪 Тестирование
### 1. Проверка подключения
```bash
cd backend
python manage.py shell
```
```python
from eusrr_backend.celery import debug_task
result = debug_task.delay()
print(f"Task ID: {result.id}")
print(f"Status: {result.status}")
```
### 2. Проверка через Django admin
- Перейти в админку: http://localhost:8000/admin/
- Найти раздел "Periodic Tasks" (от django-celery-beat)
- Создать тестовую задачу
---
## 📊 Мониторинг
### Через Flower (рекомендуется)
- URL: http://localhost:5555
- Видны: активные задачи, история, графики
### Через Redis CLI
```bash
redis-cli
> KEYS *celery*
> LLEN celery
> LRANGE celery 0 10
```
### Через Django shell
```python
from django_celery_results.models import TaskResult
TaskResult.objects.all()
```
---
## 🛑 Остановка
```bash
# Ctrl+C в каждом окне терминала (worker, beat, flower)
# Или через pkill (Linux/Mac)
pkill -f "celery worker"
pkill -f "celery beat"
pkill -f "celery flower"
```
### Остановка Redis
```bash
# Docker
docker stop redis
# Systemd
sudo systemctl stop redis
# WSL
redis-cli shutdown
```
---
## 🐛 Отладка проблем
### Проблема: Worker не запускается
**Решение:**
1. Проверьте Redis: `redis-cli ping`
2. Проверьте `.env` файл
3. Убедитесь что виртуальное окружение активно
4. Проверьте логи: `celery -A eusrr_backend worker -l debug`
### Проблема: Задачи не выполняются
**Решение:**
1. Проверьте очереди: `redis-cli LLEN celery`
2. Проверьте Flower: http://localhost:5555
3. Проверьте worker logs
4. Убедитесь что worker запущен с нужной очередью
### Проблема: Redis connection refused
**Решение:**
1. Убедитесь что Redis запущен
2. Проверьте порт: `netstat -an | grep 6379`
3. Проверьте CELERY_BROKER_URL в .env
---
## 📝 Полезные команды
### Celery
```bash
# Очистить очередь
celery -A eusrr_backend purge
# Инспекция workers
celery -A eusrr_backend inspect active
celery -A eusrr_backend inspect stats
# Список зарегистрированных задач
celery -A eusrr_backend inspect registered
```
### Redis
```bash
# Очистить все данные Celery
redis-cli KEYS "celery*" | xargs redis-cli DEL
# Мониторинг в реальном времени
redis-cli MONITOR
```
---
## 🔄 Development workflow
### Типичный процесс работы:
**Терминал 1: Django**
```bash
.venv/Scripts/python manage.py runserver
```
**Терминал 2: Celery Worker**
```bash
celery -A eusrr_backend worker -l info
```
**Терминал 3: Celery Beat** (если нужны периодические задачи)
```bash
celery -A eusrr_backend beat -l info
```
**Терминал 4: Flower** (опционально)
```bash
celery -A eusrr_backend flower
```
**Терминал 5: Redis** (если запускали вручную)
```bash
redis-server
```
---
## 🎯 Следующие шаги
После успешного запуска:
1. ✅ Создать первую задачу в `notifications/tasks.py`
2. ✅ Протестировать отправку уведомления через Celery
3. ✅ Измерить улучшение производительности
4. ✅ Настроить периодические задачи
5. ✅ Настроить мониторинг в production
---
## 📚 Документация
- Celery: https://docs.celeryq.dev/
- Flower: https://flower.readthedocs.io/
- Django Celery Beat: https://django-celery-beat.readthedocs.io/
- Redis: https://redis.io/docs/

View File

@ -0,0 +1,280 @@
# План интеграции Celery в проект EUSRR
## 📋 Текущее состояние
- ✅ Celery 5.5.3 установлен в requirements.txt
- ❌ Нет конфигурации Celery
- ❌ Нет celery.py в проекте
- ❌ Нет workers
- ⚠️ documents/tasks.py - закомментирован код с @shared_task
## 🎯 Цели интеграции
### Основные задачи:
1. **Уведомления** - перенести отправку Email/Telegram/Push в фон
2. **Документы** - отложенная отправка уведомлений для массовых рассылок
3. **Периодические задачи** - напоминания, очистка, синхронизация
### Метрики успеха:
- Скорость отклика API увеличится в 3-5 раз
- Возможность обработки 1000+ уведомлений без блокировки
- Мониторинг выполнения задач
---
## 🏗️ Архитектура решения
### Компоненты:
```
┌─────────────────┐
│ Django API │ ← HTTP requests
└────────┬────────┘
│ создает задачи
↓
┌─────────────────┐
│ Redis (Broker) │ ← очередь задач
└────────┬────────┘
│ получает задачи
↓
┌─────────────────┐
│ Celery Worker │ ← выполняет задачи
└────────┬────────┘
│ результаты
↓
┌─────────────────┐
│ Redis (Backend)│ ← хранит результаты
└─────────────────┘
```
### Очереди (Queues):
1. **default** - обычные задачи
2. **notifications** - уведомления (высокий приоритет)
3. **emails** - отправка email (низкий приоритет)
4. **periodic** - периодические задачи (cron)
---
## 📝 План реализации
### Этап 1: Базовая настройка (День 1)
**1.1. Создать структуру Celery**
- [x] `backend/eusrr_backend/celery.py` - конфигурация Celery
- [x] Обновить `backend/eusrr_backend/__init__.py` - автозагрузка
- [x] Добавить Redis в settings.py
**1.2. Настроить Redis**
- [ ] Установить Redis локально или через Docker
- [ ] Настроить CELERY_BROKER_URL
- [ ] Настроить CELERY_RESULT_BACKEND
**1.3. Тестовая задача**
- [ ] Создать простую test_task
- [ ] Запустить worker
- [ ] Проверить выполнение
---
### Этап 2: Уведомления (День 2)
**2.1. Перенести отправку уведомлений**
- [ ] `notifications/tasks.py` - создать задачи
- [ ] `send_notification_async(notification_id)`
- [ ] `send_email_notification(notification_id)`
- [ ] `send_telegram_notification(notification_id)`
- [ ] `send_push_notification(notification_id)`
**2.2. Обновить NotificationService**
- [ ] Изменить `send_notification()` - использовать Celery
- [ ] Добавить параметр `use_celery=True`
- [ ] Fallback на синхронную отправку при ошибке
**2.3. Тестирование**
- [ ] Проверить отправку уведомлений через Celery
- [ ] Измерить время отклика API (было/стало)
---
### Этап 3: Документы (День 3)
**3.1. Раскомментировать tasks.py**
- [ ] documents/tasks.py - восстановить функционал
- [ ] Адаптировать под новую архитектуру
**3.2. Массовые рассылки**
- [ ] Батчинг: группировать уведомления по N штук
- [ ] Прогресс-бар для админа (сколько отправлено)
---
### Этап 4: Периодические задачи (День 4)
**4.1. Celery Beat**
- [ ] Настроить celery beat
- [ ] Создать schedule в settings.py
**4.2. Задачи**
- [ ] Напоминания о документах (раз в день)
- [ ] Очистка старых уведомлений (раз в неделю)
- [ ] Синхронизация LDAP (если нужно)
- [ ] Backup базы данных (раз в сутки)
---
### Этап 5: Мониторинг (День 5)
**5.1. Flower**
- [ ] Установить Flower
- [ ] Настроить доступ
- [ ] Дашборд для мониторинга
**5.2. Логирование**
- [ ] Structured logging для задач
- [ ] Алерты при сбоях (опционально - Sentry)
**5.3. Метрики**
- [ ] Время выполнения задач
- [ ] Количество успешных/неуспешных
- [ ] Размер очереди
---
## ⚙️ Конфигурация
### Redis settings (в .env):
```env
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
CELERY_BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/1
```
### Celery settings (в settings.py):
```python
CELERY_BROKER_URL = os.getenv('CELERY_BROKER_URL', 'redis://localhost:6379/0')
CELERY_RESULT_BACKEND = os.getenv('CELERY_RESULT_BACKEND', 'redis://localhost:6379/1')
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Europe/Moscow'
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60 # 30 минут
```
### Запуск worker:
```bash
# Development
celery -A eusrr_backend worker -l info -Q default,notifications,emails
# Production (with autoscale)
celery -A eusrr_backend worker -l info --autoscale=10,3
```
### Запуск beat:
```bash
celery -A eusrr_backend beat -l info
```
### Запуск Flower:
```bash
celery -A eusrr_backend flower --port=5555
```
---
## 🚀 Deployment
### Docker Compose (пример):
```yaml
services:
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
celery_worker:
build: .
command: celery -A eusrr_backend worker -l info
volumes:
- .:/app
depends_on:
- redis
- db
environment:
- CELERY_BROKER_URL=redis://redis:6379/0
celery_beat:
build: .
command: celery -A eusrr_backend beat -l info
volumes:
- .:/app
depends_on:
- redis
- db
flower:
build: .
command: celery -A eusrr_backend flower --port=5555
ports:
- "5555:5555"
depends_on:
- redis
- celery_worker
```
---
## ⚠️ Риски и меры минимизации
### Риск 1: Redis недоступен
**Решение:** Fallback на синхронную обработку
### Риск 2: Worker умер
**Решение:** Supervisor/systemd для автоперезапуска
### Риск 3: Очередь переполнена
**Решение:** Rate limiting + мониторинг через Flower
### Риск 4: Долгие задачи блокируют
**Решение:** Разделение на очереди + task time limit
---
## 📊 Метрики для отслеживания
**До внедрения:**
- Время отправки сообщения в групповой чат: ~500ms
- Время создания документа с 100 получателями: ~10s
**После внедрения (цель):**
- Время отправки сообщения: <50ms
- Время создания документа: <200ms
- Уведомления в фоне: не блокируют API
---
## ✅ Критерии готовности к продакшену
- [ ] Все задачи выполняются стабильно
- [ ] Worker автоматически перезапускается при сбое
- [ ] Flower доступен для мониторинга
- [ ] Логирование настроено
- [ ] Redis backup настроен
- [ ] Тестирование под нагрузкой пройдено
- [ ] Документация для деплоя готова
---
## 🔄 Откат изменений
Если что-то пойдет не так:
1. Вернуться к синхронной отправке в NotificationService
2. Закомментировать Celery задачи
3. Переключиться обратно на ветку master
**Git:** `git checkout master && git branch -D feature/celery-integration`