Какой вид многопоточности будешь использовать для рассылки писем?

Выбор многопоточности для рассылки писем

Для рассылки писем я бы использовал асинхронное программирование (asyncio), а не потоки. Вот почему и как это работает:

Почему именно asyncio, а не threading?

Threading (потоки):

• GIL (Global Interpreter Lock) блокирует параллельное выполнение Python кода
• Подходит только для I/O-bound операций
• Сложнее отлаживать и возникают race conditions
• Дороже по памяти (каждый поток занимает ~8MB)

Asyncio (асинхронность):

• Не подвержен GIL, так как работает в одном потоке
• Идеален для I/O-bound задач (отправка писем — это I/O)
• Простой контроль потока выполнения через await
• Экономнее памяти (можно запустить тысячи корутин)
• Естественный способ работы с сетью и почтой

Реализация с asyncio

import asyncio
import aiosmtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

class EmailSender:
    def __init__(self, smtp_host: str, smtp_port: int):
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
    
    async def send_email(self, to: str, subject: str, body: str) -> bool:
        """Отправить одно письмо асинхронно"""
        try:
            async with aiosmtplib.SMTP(hostname=self.smtp_host, port=self.smtp_port) as smtp:
                message = MIMEMultipart()
                message["From"] = "noreply@example.com"
                message["To"] = to
                message["Subject"] = subject
                message.attach(MIMEText(body, "html"))
                
                await smtp.send_message(message)
                return True
        except Exception as e:
            print(f"Ошибка отправки письма {to}: {e}")
            return False
    
    async def send_bulk(
        self, 
        recipients: list[dict],  # [{"email": "...", "subject": "...", "body": "..."}, ...]
        max_concurrent: int = 10
    ) -> dict:
        """Отправить письма с ограничением на одновременные запросы"""
        semaphore = asyncio.Semaphore(max_concurrent)
        
        async def send_with_limit(recipient):
            async with semaphore:
                return await self.send_email(
                    to=recipient["email"],
                    subject=recipient["subject"],
                    body=recipient["body"]
                )
        
        results = await asyncio.gather(
            *[send_with_limit(r) for r in recipients],
            return_exceptions=True
        )
        
        return {
            "total": len(recipients),
            "sent": sum(1 for r in results if r is True),
            "failed": sum(1 for r in results if r is not True)
        }

# Использование
async def main():
    sender = EmailSender(smtp_host="smtp.gmail.com", smtp_port=587)
    recipients = [
        {"email": "user1@example.com", "subject": "Привет", "body": "Содержание письма"},
        {"email": "user2@example.com", "subject": "Привет", "body": "Содержание письма"},
        # ... тысячи писем
    ]
    
    result = await sender.send_bulk(recipients, max_concurrent=10)
    print(f"Результат: {result}")

asyncio.run(main())

Альтернатива: Celery (для очередей)

Для enterprise-решений использую Celery с Redis/RabbitMQ:

from celery import Celery, group
from celery.result import GroupResult

app = Celery("mail_service", broker="redis://localhost:6379")

@app.task(bind=True, max_retries=3)
def send_email_task(self, to: str, subject: str, body: str):
    """Задача для отправки письма с повторами"""
    try:
        # SMTP логика здесь
        return {"status": "sent", "email": to}
    except Exception as exc:
        # Повтор через 60 сек
        raise self.retry(exc=exc, countdown=60)

def send_bulk_emails(recipients: list[dict]):
    """Поставить задачи в очередь"""
    job = group(
        send_email_task.s(
            to=r["email"],
            subject=r["subject"],
            body=r["body"]
        )
        for r in recipients
    )
    return job.apply_async()

Ключевые моменты при выборе

Используй asyncio если:

• Письма отправляются синхронно в одном процессе
• Нужна простота развертывания
• < 10,000 писем в день
• Письма некритичны по надежности

Используй Celery если:

• Критична надежность доставки
• Нужны повторы при ошибках
• Письма отправляются из разных мест приложения
• Нужна обработка в фоне

• 10,000 писем в день

Бонус: Использование thread pool для блокирующих операций

Если необходимо вызвать синхронную библиотеку SMTP:

import asyncio
from concurrent.futures import ThreadPoolExecutor
import smtplib

loop = asyncio.get_event_loop()
executor = ThreadPoolExecutor(max_workers=5)

async def send_email_threaded(to: str, subject: str):
    def _send():
        # Синхронный SMTP код
        with smtplib.SMTP("smtp.gmail.com") as smtp:
            smtp.send_message(...)
    
    return await loop.run_in_executor(executor, _send)

Вывод: Для рассылки писем asyncio + aiosmtplib — идеальный выбор благодаря эффективности, простоте и отсутствию GIL проблем.