एक Django आवेदन से asyncio और aiohttp का उपयोग कर ईमेल भेजें

सभी को नमस्कार!

मैं Ostrovok.ru में एक अधिसूचना सेवा का विकास और समर्थन कर रहा हूं। सेवा Python3 और Django में लिखी गई है। लेन-देन पत्र, धक्का और संदेशों के अलावा, यह सेवा उन उपयोगकर्ताओं के लिए भी काम करती है जो वाणिज्यिक ऑफ़र के बड़े पैमाने पर मेलिंग करते हैं (स्पैम नहीं! मुझ पर भरोसा करें, सदस्यता रद्द करें सदस्यता से बेहतर काम करते हैं) उन उपयोगकर्ताओं के लिए जिन्होंने इस पर सहमति दी है। समय के साथ, सक्रिय प्राप्तकर्ताओं का आधार एक मिलियन से अधिक पते तक बढ़ गया, जिसके लिए मेल सेवा तैयार नहीं थी। मैं इस बारे में बात करना चाहता हूं कि कैसे नए पायथन फीचर्स ने बड़े पैमाने पर मेलिंग को तेज करने और संसाधनों को बचाने के लिए संभव बनाया और उनके साथ काम करते समय हमें किन समस्याओं से जूझना पड़ा।

स्रोत कार्यान्वयन

प्रारंभ में, बड़े पैमाने पर मेल को सबसे सरल तरीके से लागू किया गया था: प्रत्येक प्राप्तकर्ता के लिए, एक कार्य एक कतार में रखा गया था जिसे 60 मास श्रमिकों में से एक द्वारा लिया गया था (हमारी कतार की एक विशेषता यह है कि प्रत्येक कार्यकर्ता एक अलग प्रक्रिया में काम करता है), इसके लिए संदर्भ तैयार किया, टेम्पलेट प्रदान किया, भेजा एक पत्र भेजने के लिए Mailgun के लिए एक HTTP अनुरोध और पत्र भेजे जाने वाले डेटाबेस में एक रिकॉर्ड बनाया। संपूर्ण मेलिंग में 12 घंटे तक का समय लगा, प्रत्येक कार्यकर्ता से लगभग 0.3 अक्षर प्रति सेकंड भेजने और छोटे अभियानों के मेलिंग को अवरुद्ध करने के लिए।

अतुल्यकालिक समाधान

त्वरित रूपरेखा ने दिखाया कि अधिकांश समय कार्यकर्ता मेलगुन के साथ संबंध स्थापित करने पर खर्च करते हैं, इसलिए हमने प्रत्येक कार्यकर्ता के लिए चंक करके समूह कार्यों को शुरू किया। श्रमिकों ने मेलगुन के साथ एक कनेक्शन का उपयोग करना शुरू कर दिया, जिससे मेलिंग का समय 9 घंटे तक कम करने की अनुमति मिली, प्रत्येक कार्यकर्ता को औसतन 0.5 अक्षर प्रति सेकंड भेजा गया। बाद में फिर से प्रोफाइलिंग से पता चला कि नेटवर्क के साथ काम करने में अभी भी अधिकांश समय लगता है, जिसने हमें एसिंको का उपयोग करने के लिए प्रेरित किया।

सभी प्रसंस्करण को एसिंसीओ चक्र में डालने से पहले, हमें कई समस्याओं के समाधान के माध्यम से सोचना था:

1. Django ORM अभी तक asyncio के साथ काम करने में सक्षम नहीं है, हालांकि, यह क्वेरी निष्पादन के दौरान GIL को रिलीज़ करता है। इसका मतलब है कि डेटाबेस प्रश्नों को एक अलग थ्रेड में निष्पादित किया जा सकता है और मुख्य लूप को ब्लॉक नहीं किया जा सकता है।
2. आईओएचटीटीपी के वर्तमान संस्करणों को पायथन संस्करण 3.6 और उच्चतर की आवश्यकता होती है, जो कार्यान्वयन के समय में डॉकटर छवि को अपडेट करने की आवश्यकता होती है। एनोहॉटप और पायथन 3.5 के पुराने संस्करणों पर प्रयोग से पता चला कि इन संस्करणों पर भेजने की गति नए संस्करणों की तुलना में बहुत कम है, और अनुक्रमिक भेजने के लिए तुलनीय है।
3. बड़ी मात्रा में एसिंसीओ कोरुटिन को स्टोर करने से सभी मेमोरी का खर्च होता है। इसका मतलब यह है कि सभी कोरआउट को पत्रों के लिए पूर्व-तैयार करना असंभव है और उन्हें संसाधित करने के लिए एक चक्र का कारण बनता है, डेटा तैयार करना आवश्यक है क्योंकि आप पहले से ही गठित पत्र भेजते हैं।

सभी विशेषताओं को ध्यान में रखते हुए, हम अपने प्रत्येक कर्मचारी को थ्रेडपूल पैटर्न की समानता के साथ हमारे एसिनोसियो चक्र के अंदर बनाएंगे, जिसमें निम्न शामिल हैं:

• एक या एक से अधिक उत्पादकों के साथ एक अलग थ्रेड में Django ORM डेटाबेस के साथ asyncio.ThreadPoolExecutor के माध्यम से काम कर रहा है। निर्माता छोटे बैचों में डेटा अधिग्रहण अनुरोधों को एकत्र करने की कोशिश करता है, जिन्जा 2 के माध्यम से प्राप्त डेटा के लिए टेम्पलेट्स प्रदान करता है और कार्य कतार में भेजने के लिए डेटा जोड़ता है।

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) 

• अक्षरों के कई सौ प्रेषक - एसिंसीओ कोरटाइन, जो एक अंतहीन चक्र में कार्य कतार से डेटा पढ़ते हैं, उनमें से प्रत्येक के लिए नेटवर्क अनुरोध भेजते हैं और रिपोर्ट कतार में परिणाम (प्रतिक्रिया, या अपवाद) डालते हैं।
  

 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() 

• एक या एक से अधिक कार्यकर्ता जो रिपोर्ट कतार से डेटा समूह करते हैं और एक अनुरोध के साथ बल्क डेटाबेस को एक पत्र भेजने के परिणाम पर जानकारी डालते हैं।

 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() 

• कार्य कतार, जो asyncio.Queue का एक उदाहरण है, तत्वों की अधिकतम संख्या द्वारा सीमित है ताकि निर्माता इसे ओवरफिल न करें, सभी मेमोरी खर्च करें।
• रिपोर्ट कतारों, भी asyncio.Queue का एक उदाहरण अधिकतम आइटम की सीमा के साथ।
• एक एसिंक्रोनस विधि जो कतारों, श्रमिकों को बनाती है, और उन्हें रोककर वितरण पूरा करती है।

 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() 

• समकालिक कोड जो लूप बनाता है और वितरण शुरू करता है।

 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() 

इस समाधान को लागू करने के बाद, एक ही मेलिंग वॉल्यूम और 12 शामिल श्रमिकों के साथ सामूहिक डाक भेजने का समय एक घंटे तक कम हो गया था। यही है, प्रत्येक कार्यकर्ता प्रति सेकंड 20-25 पत्र भेजता है, जो मूल समाधान की तुलना में 50-80 गुना अधिक उत्पादक है। श्रमिकों की स्मृति खपत प्रारंभिक स्तर पर बनाए रखी गई थी, प्रोसेसर लोड थोड़ा बढ़ गया, नेटवर्क उपयोग कई बार बढ़ गया, जो अपेक्षित प्रभाव है। डेटाबेस से कनेक्शन की संख्या में भी वृद्धि हुई है, क्योंकि कार्यकर्ता-निर्माता और श्रमिकों के प्रत्येक प्रवाह, जो रिपोर्ट को सहेजते हैं, डेटाबेस के साथ सक्रिय रूप से काम करते हैं। एक ही समय में, मुफ्त कार्यकर्ता छोटे मेलिंग भेज सकते हैं जबकि एक बड़े पैमाने पर अभियान भेजा जाता है।



सभी लाभों के बावजूद, इस तरह के कार्यान्वयन में कई कठिनाइयाँ होती हैं जिन्हें ध्यान में रखा जाना चाहिए:

1. त्रुटियों को संभालते समय सावधानी बरतनी चाहिए। एक अनियंत्रित अपवाद कार्यकर्ता को समाप्त कर सकता है, जिससे अभियान स्थिर हो जाएगा।
2. जब भेजना पूरा हो जाता है, तो यह आवश्यक है कि उन प्राप्तकर्ताओं पर रिपोर्ट न खोएं जिन्होंने अंत तक चंक पूरा नहीं किया है, और उन्हें डेटाबेस में सहेजें।
3. अभियानों को फिर से शुरू करने से जबरन रोकने का तर्क अधिक जटिल हो रहा है, क्योंकि भेजने वाले श्रमिकों को रोकने के बाद, यह तुलना करना आवश्यक है कि कौन से प्राप्तकर्ता पत्र भेजे गए थे और कौन से नहीं थे।
4. कुछ समय बाद, Mailgun सपोर्ट स्टाफ ने हमसे संपर्क किया और हमें भेजने की गति को धीमा करने के लिए कहा, क्योंकि मेल सेवाओं ने अस्थायी रूप से ईमेल को अस्वीकार करना शुरू कर दिया था यदि उनके भेजने की आवृत्ति थ्रेशोल्ड मान से अधिक हो। श्रमिकों की संख्या कम करके ऐसा करना आसान है।
5. अगर एसेंस-डिमांड ऑपरेशंस को पूरा करने के लिए लेटर भेजने के कुछ चरणों में एसिंको का इस्तेमाल करना संभव नहीं होगा। Jinja2 का उपयोग करते हुए टेम्प्लेटिंग टेम्प्लेट एक बहुत ही संसाधन-गहन संचालन नहीं है और इसका व्यावहारिक रूप से भेजने की गति पर कोई प्रभाव नहीं पड़ता है।
6. मेलिंग सूचियों के लिए एसिंको का उपयोग करके वितरण कतार संचालकों को अलग-अलग प्रक्रियाओं द्वारा शुरू करने की आवश्यकता होती है।

मुझे आशा है कि हमारा अनुभव आपके लिए उपयोगी होगा! यदि आपके कोई प्रश्न या विचार हैं, तो टिप्पणियों में लिखें!