Envoyer des e-mails en utilisant asyncio et aiohttp à partir d'une application Django

Bonjour à tous!

Je développe et supporte un service de notification sur Ostrovok.ru . Le service est écrit en Python3 et Django. En plus des lettres, push et messages transactionnels, le service entreprend également des tâches d'envoi massif d'offres commerciales (pas de spam! Faites-moi confiance, les désabonnements fonctionnent mieux que les abonnements) aux utilisateurs qui ont donné leur accord. Au fil du temps, la base de destinataires actifs est passée à plus d'un million d'adresses, pour lesquelles le service de messagerie n'était pas prêt. Je veux parler de la façon dont les nouvelles fonctionnalités de Python ont permis d'accélérer les envois de masse et d'économiser des ressources et des problèmes que nous avons dû traiter lors de leur utilisation.

Implémentation de la source

Initialement, les publipostages de masse ont été mis en œuvre de la manière la plus simple: pour chaque destinataire, une tâche a été placée dans une file d'attente qui a été prise par l'un des 60 travailleurs de masse (une caractéristique de nos files d'attente est que chaque travailleur travaille dans un processus séparé), a préparé le contexte pour cela, rendu le modèle, envoyé Une requête HTTP à Mailgun pour envoyer une lettre et créé un enregistrement dans la base de données que la lettre a été envoyée. L'envoi complet a duré jusqu'à 12 heures, envoyant environ 0,3 lettre par seconde de chaque travailleur et bloquant l'envoi de petites campagnes.

Solution asynchrone

Le profilage rapide a montré que la plupart du temps que les travailleurs passent à établir des connexions avec Mailgun, nous avons donc commencé à regrouper les tâches en blocs, par bloc pour chaque travailleur. Les travailleurs ont commencé à utiliser une connexion avec Mailgun, ce qui a permis de réduire le temps d'envoi à 9 heures, envoyant à chaque travailleur une moyenne de 0,5 lettres par seconde. Le profilage ultérieur a de nouveau montré que travailler avec le réseau prend toujours la plupart du temps, ce qui nous a incités à utiliser asyncio.

Avant de mettre tout le traitement dans le cycle asyncio, nous avons dû réfléchir à une solution à un certain nombre de problèmes:

1. Django ORM n'est pas encore capable de fonctionner avec asyncio, cependant, il libère le GIL pendant l'exécution de la requête. Cela signifie que les requêtes de base de données peuvent être exécutées dans un thread séparé et ne pas bloquer la boucle principale.
2. Les versions actuelles d'aiohttp nécessitent Python version 3.6 et supérieure, ce qui, au moment de l'implémentation, nécessitait la mise à jour de l'image docker. Les expériences sur les anciennes versions d'aiohttp et Python 3.5 ont montré que la vitesse d'envoi sur ces versions est beaucoup plus faible que sur les versions plus récentes, et est comparable à l'envoi séquentiel.
3. Stocker une grande quantité de corutine asyncio conduit rapidement à la dépense de toute la mémoire. Cela signifie qu'il est impossible de pré-préparer toutes les coroutines pour les lettres et de provoquer un cycle pour les traiter, il est nécessaire de préparer les données lorsque vous envoyez des lettres déjà formées.

Compte tenu de toutes les fonctionnalités, nous allons créer à l'intérieur de chacun de nos travailleurs notre cycle asyncio avec la similitude du modèle ThreadPool, composé de:

• Un ou plusieurs producteurs travaillant avec une base de données via Django ORM dans un thread séparé via asyncio.ThreadPoolExecutor. Le fabricant essaie d'agréger les demandes d'acquisition de données en petits lots, rend les modèles des données reçues via Jinja2 et ajoute les données à envoyer à la file d'attente des tâches.

def get_campaign_send_data(ids: Iterable[int]) -> Iterable[Mapping[str, Any]]: """    ,     Django ORM   .""" return [{'id': id} for id in ids] async def mail_campaign_producer(ids: Iterable[int], task_queue: asyncio.Queue) -> None: """           ,    .      ,     ThreadPoolExecutor. """ loop = asyncio.get_event_loop() total = len(ids) for subchunk_start in range(0, total, PRODUCER_SUBCHUNK_SIZE): subchunk_ids = ids[subchunk_start : min(subchunk_start + PRODUCER_SUBCHUNK_SIZE, total)] send_tasks = await loop.run_in_executor(None, get_campaign_send_data, subchunk_ids) for task in send_tasks: await task_queue.put(task) 

• Plusieurs centaines d'expéditeurs de lettres - des coroutines asyncio, qui, dans un cycle sans fin, lisent les données de la file d'attente des tâches, envoient des demandes réseau pour chacune d'entre elles et mettent le résultat (réponse ou exception) dans la file d'attente des rapports.
  

 async def send_mail(data: Mapping[str, Any], session: aiohttp.ClientSession) -> Union[Mapping[str, Any], Exception]: """    .""" async with session.post(REQUEST_URL, data=data) as response: if response.status_code != 200: raise Exception return data async def mail_campaign_sender( task_queue: asyncio.Queue, result_queue: asyncio.Queue, session: aiohttp.ClientSession ) -> None: """        .     task_done,    ,   . """ while True: try: task_data = await task_queue.get() result = await send_mail(task_data, session) await result_queue.put(result) except asyncio.CancelledError: #     raise except Exception as exception: #     await result_queue.put(exception) finally: task_queue.task_done() 

• Un ou plusieurs travailleurs qui regroupent les données de la file d'attente de rapports et mettent des informations sur le résultat de l'envoi d'une lettre à la base de données en masse avec une demande.

 def process_campaign_results(results: Iterable[Union[Mapping[str, Any], Exception]]) -> None: """  :         """ pass async def mail_campaign_reporter(task_queue: asyncio.Queue, result_queue: asyncio.Queue) -> None: """          ThreadPoolExecutor,        . """ loop = asyncio.get_event_loop() results_chunk = [] while True: try: results_chunk.append(await result_queue.get()) if len(results_chunk) >= REPORTER_BATCH_SIZE: await loop.run_in_executor(None, process_campaign_results, results_chunk) results_chunk.clear() except asyncio.CancelledError: await loop.run_in_executor(None, process_campaign_results, results_chunk) results_chunk.clear() raise finally: result_queue.task_done() 

• La file d'attente des tâches, qui est une instance de asyncio.Queue, limitée par le nombre maximal d'éléments afin que le fabricant ne la remplisse pas trop, dépensant toute la mémoire.
• Reportez les files d'attente, également une instance de asyncio.Queue avec une limite sur le nombre maximal d'éléments.
• Une méthode asynchrone qui crée des files d'attente, des travailleurs et termine la distribution en les arrêtant.

 async def send_mail_campaign( recipient_ids: Iterable[int], session: aiohttp.ClientSession, loop: asyncio.AbstractEventLoop = None ) -> None: """       .    ,       . """ executor = ThreadPoolExecutor(max_workers=PRODUCERS_COUNT + 1) loop = loop or asyncio.get_event_loop() loop.set_default_executor(executor) task_queue = asyncio.Queue(maxsize=2 * SENDERS_COUNT, loop=loop) result_queue = asyncio.Queue(maxsize=2 * SENDERS_COUNT, loop=loop) producers = [ asyncio.ensure_future(mail_campaign_producer(recipient_ids, task_queue)) for _ in range(PRODUCERS_COUNT) ] consumers = [ asyncio.ensure_future(mail_campaign_sender(task_queue, result_queue, session)) for _ in range(SENDERS_COUNT) ] reporter = asyncio.ensure_future(mail_campaign_reporter(task_queue, result_queue)) # ,      done, _ = await asyncio.wait(producers) #    ,   await task_queue.join() while consumers: consumers.pop().cancel() #    ,     await result_queue.join() reporter.cancel() 

• Le code synchrone qui crée la boucle et démarre la distribution.

 async def close_session(future: asyncio.Future, session: aiohttp.ClientSession) -> None: """  ,    .  aiohttp      . """ await asyncio.wait([future]) await asyncio.sleep(0.250) await session.close() def mail_campaign_send_chunk(recipient_ids: Iterable[int]) -> None: """     .   ,  asyncio     . """ loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # Session connector = aiohttp.TCPConnector(limit_per_host=0, limit=0) session = aiohttp.ClientSession( connector=connector, auth=aiohttp.BasicAuth('api', API_KEY), loop=loop, read_timeout=60 ) send_future = asyncio.ensure_future(send_mail_campaign(recipient_ids, session, loop=loop)) cleanup_future = asyncio.ensure_future(close_session(send_future, session)) loop.run_until_complete(asyncio.wait([send_future, cleanup_future])) loop.close() 

Après la mise en œuvre de cette solution, le délai d'envoi de mailings en masse a été réduit à une heure avec les mêmes volumes d'envoi et 12 collaborateurs impliqués. Autrement dit, chaque travailleur envoie 20 à 25 lettres par seconde, ce qui est 50 à 80 fois plus productif que la solution d'origine. La consommation de mémoire des travailleurs a été maintenue au niveau initial, la charge du processeur a légèrement augmenté, l'utilisation du réseau a augmenté plusieurs fois, ce qui est l'effet attendu. Le nombre de connexions à la base de données a également augmenté, car chacun des flux de travailleurs-producteurs et de travailleurs qui enregistrent des rapports travaille activement avec la base de données. Dans le même temps, les travailleurs libres peuvent envoyer de petits mailings pendant qu'une campagne de masse est envoyée.



Malgré tous les avantages, une telle mise en œuvre présente un certain nombre de difficultés à prendre en compte:

1. Des précautions doivent être prises lors de la manipulation des erreurs. Une exception non gérée peut mettre fin au travailleur, provoquant le gel de la campagne.
2. Lorsque l'envoi est terminé, il est nécessaire de ne pas perdre les rapports sur les destinataires qui n'ont pas terminé le bloc à la fin et de les enregistrer dans la base de données.
3. La logique de l'arrêt forcé de la reprise des campagnes devient de plus en plus compliquée, car après l'arrêt des travailleurs expéditeurs, il est nécessaire de comparer quels destinataires ont été envoyés et lesquels ne l'ont pas été.
4. Après un certain temps, le personnel d'assistance de Mailgun nous a contactés et nous a demandé de ralentir la vitesse d'envoi, car les services de messagerie ont commencé à rejeter temporairement les e-mails si la fréquence de leur envoi dépasse la valeur seuil. Ceci est facile à faire en réduisant le nombre de travailleurs.
5. Il ne serait pas possible d'utiliser asyncio si certaines des étapes de l'envoi de lettres effectuaient des opérations exigeantes en termes de processeur. Le rendu des modèles utilisant jinja2 s'est avéré être une opération peu gourmande en ressources et n'a pratiquement aucun effet sur la vitesse d'envoi.
6. L'utilisation d'asyncio pour les listes de diffusion nécessite que les gestionnaires de files d'attente de distribution soient démarrés par des processus distincts.

J'espère que notre expérience vous sera utile! Si vous avez des questions ou des idées, écrivez dans les commentaires!