Asyncio coroutine पैटर्न: बाहर प्रतीक्षारत

अनुवादक का पूर्वानुमान:
एक बार फिर अजगर असिनसियो के साथ काम करने के दौरान एक रेक पर कदम रखते हुए, मैं ड्राई डॉक्यूमेंटेशन के लिए कुछ और सुखद खोजने के लिए इंटरनेट पर गया। मुझे येर डायज़ "असिनसियो कोराउटाइन पैटर्न: बियॉन्ड वेट" का एक लेख आया, जिसमें लेखक ने बहुत ही रोचक तरीके से एसिनसियो के उपयोग की जांच की और कुछ ट्रिक्स शेयर किए। चूंकि मुझे रूसी में कुछ भी पूरा नहीं मिला, इसलिए मैंने इसका अनुवाद करने का फैसला किया।

Asyncio एक अजगर प्रोग्रामर का प्रतिस्पर्धी सपना है: आप सिंक्रोनस पर कोडिंग लिखते हैं और पायथन को आराम करने देते हैं। यह एंटी-ग्रेविटी लाइब्रेरी का एक और आयात है: एंटी-ग्रेविटी का आयात

वास्तव में, ऐसा बिल्कुल नहीं है, समवर्ती प्रोग्रामिंग एक कठिन काम है और, जबकि कोरटाइन हमें कॉलबैक नरक से बचने की अनुमति देते हैं, जो आपको काफी दूर ले जा सकता है, आपको अभी भी कार्य बनाने, परिणाम प्राप्त करने और अपवादों को पकड़ने के बारे में सोचने की आवश्यकता है। यह दुखद है।

अच्छी खबर यह है कि यह सब asyncio में संभव है। बुरी खबर यह है कि यह हमेशा स्पष्ट नहीं है कि क्या गलत है और इसे कैसे ठीक किया जाए। नीचे कुछ पैटर्न हैं जो मैंने एसिंको के साथ काम करते समय खोजे थे।

शुरू करने से पहले:

मैंने अतुल्यकालिक HTTP अनुरोधों और हैकर न्यूज़ एपीआई को निष्पादित करने के लिए अपने पसंदीदा aiohttp लाइब्रेरी का उपयोग किया क्योंकि यह एक सरल और प्रसिद्ध साइट है जो एक परिचित उपयोग परिदृश्य का अनुसरण करती है। मेरे पिछले लेख की प्रतिक्रिया के बाद, मैंने पायथन 3.5 में पेश किए गए async / प्रतीक्षा सिंटैक्स का भी उपयोग किया। मैंने यह माना कि पाठक यहाँ वर्णित विचारों से परिचित था। और अंततः, सभी उदाहरण इस लेख के गिटहब भंडार में उपलब्ध हैं।

ठीक है, चलो शुरू हो जाओ!

पुनरावर्ती कोराटाइन

कार्यों को बनाना और चलाना asyncio में तुच्छ है। ऐसे कार्यों के लिए, API में AbstractEventLoop वर्ग के साथ-साथ लाइब्रेरी में कार्य करने के कई तरीके शामिल हैं। लेकिन आम तौर पर आप इन कार्यों से परिणामों को संयोजित करना चाहते हैं और उन्हें किसी तरह से संसाधित करते हैं, और पुनरावृत्ति इस योजना का एक बड़ा उदाहरण है, और अन्य प्रतिस्पर्धी साधनों की तुलना में कोरटाइन की सादगी को भी दर्शाता है।

Asyncio का उपयोग करने के लिए एक सामान्य मामला कुछ प्रकार के वेब क्रॉलर बनाना है। कल्पना कीजिए कि हम सिर्फ HackerNews पर जांच करने में व्यस्त हैं, या हो सकता है कि आप एक अच्छे होलीवर की तरह हों, इसलिए आप एक ऐसी प्रणाली को लागू करना चाहते हैं जो किसी विशिष्ट HN पोस्ट के लिए टिप्पणियों की संख्या को पुनर्प्राप्त करती है और यदि यह सीमा से ऊपर है तो आपको सूचित करती है। आपने थोड़ा सा गुमराह किया और HN API के लिए प्रलेखन पाया, बस आपको क्या चाहिए, लेकिन आपने दस्तावेज़ में निम्नलिखित पर ध्यान दिया:

लेख टिप्पणियों की कुल संख्या जानना चाहते हैं? पेड़ के चारों ओर जाओ और उन्हें गिनो।

कॉल स्वीकार किया गया!

"""          ,       ,      . ,         Hacker News      """ import asyncio import argparse import logging from urllib.parse import urlparse, parse_qs from datetime import datetime import aiohttp import async_timeout LOGGER_FORMAT = '%(asctime)s %(message)s' URL_TEMPLATE = "https://hacker-news.firebaseio.com/v0/item/{}.json" FETCH_TIMEOUT = 10 parser = argparse.ArgumentParser(description='Calculate the comments of a Hacker News post.') parser.add_argument('--id', type=int, default=8863, help='ID of the post in HN, defaults to 8863') parser.add_argument('--url', type=str, help='URL of a post in HN') parser.add_argument('--verbose', action='store_true', help='Detailed output') logging.basicConfig(format=LOGGER_FORMAT, datefmt='[%H:%M:%S]') log = logging.getLogger() log.setLevel(logging.INFO) fetch_counter = 0 async def fetch(session, url): """  URL   aiohttp,    JSON .     aiohttp    . """ global fetch_counter with async_timeout.timeout(FETCH_TIMEOUT): fetch_counter += 1 async with session.get(url) as response: return await response.json() async def post_number_of_comments(loop, session, post_id): """          . """ url = URL_TEMPLATE.format(post_id) now = datetime.now() response = await fetch(session, url) log.debug('{:^6} > Fetching of {} took {} seconds'.format( post_id, url, (datetime.now() - now).total_seconds())) if 'kids' not in response: #  .   return 0 #     ,    (  ) number_of_comments = len(response['kids']) #       log.debug('{:^6} > Fetching {} child posts'.format( post_id, number_of_comments)) tasks = [post_number_of_comments(loop, session, kid_id) for kid_id in response['kids']] #      results = await asyncio.gather(*tasks) #            number_of_comments += sum(results) log.debug('{:^6} > {} comments'.format(post_id, number_of_comments)) return number_of_comments def id_from_HN_url(url): """   `id` URL       None. """ parse_result = urlparse(url) try: return parse_qs(parse_result.query)['id'][0] except (KeyError, IndexError): return None if __name__ == '__main__': args = parser.parse_args() if args.verbose: log.setLevel(logging.DEBUG) post_id = id_from_HN_url(args.url) if args.url else args.id loop = asyncio.get_event_loop() with aiohttp.ClientSession(loop=loop) as session: now = datetime.now() comments = loop.run_until_complete( post_number_of_comments(loop, session, post_id)) log.info( '> Calculating comments took {:.2f} seconds and {} fetches'.format( (datetime.now() - now).total_seconds(), fetch_counter)) log.info("-- Post {} has {} comments".format(post_id, comments)) loop.close() 

अधिक विस्तृत आउटपुट के लिए "-verbose" ध्वज के साथ स्क्रिप्ट चलाने का प्रयास करने के लिए स्वतंत्र महसूस करें।

  [14:47:32] > Calculating comments took 2.23 seconds and 73 fetches [14:47:32] -- Post 8863 has 72 comments 

चलो बॉयलरप्लेट कोड को छोड़ दें और पुनरावर्ती कोरआउट में सीधे जाएं। ध्यान दें कि यह कोड लगभग पूरी तरह से पढ़ा गया है क्योंकि यह सिंक्रोनस कोड के साथ होगा।

 async def post_number_of_comments(loop, session, post_id): """          . """ url = URL_TEMPLATE.format(post_id) now = datetime.now() response = await fetch(session, url) log.debug('{:^6} > Fetching of {} took {} seconds'.format( post_id, url, (datetime.now() - now).total_seconds())) if 'kids' not in response: # base case, there are no comments return 0 #     ,    (  ) number_of_comments = len(response['kids']) #       log.debug('{:^6} > Fetching {} child posts'.format( post_id, number_of_comments)) tasks = [post_number_of_comments( loop, session, kid_id) for kid_id in response['kids']] #      results = await asyncio.gather(*tasks) #          number_of_comments += sum(results) log.debug('{:^6} > {} comments'.format(post_id, number_of_comments)) return number_of_comments 

1. पहले हम JSON को पोस्ट डेटा के साथ प्राप्त करते हैं।
2. प्रत्येक उत्तराधिकारी के आसपास पुन: भ्रमण करें।
3. अंत में, हम आधार मामले तक पहुँचेंगे और शून्य वापस करेंगे,
   जब पोस्ट का कोई फीडबैक नहीं है।
4. आधार मामले से लौटते समय, उत्तर को वर्तमान पोस्ट में जोड़ें
   वारिस की संख्या और वापसी

ब्रेट स्लेटकिन ने फैन-इन और फैन-आउट के रूप में जो वर्णन किया है, उसका यह एक बड़ा उदाहरण है , हम वारिसों से डेटा प्राप्त करने के लिए फैन-आउट हैं और प्रशंसकों ने टिप्पणियों की संख्या की गणना करने के लिए डेटा को संक्षेप में प्रस्तुत किया है

इन फैन-आउट ऑपरेशन को करने के लिए asyncio API में कुछ तरीके हैं। यहां मैं एकत्रित फ़ंक्शन का उपयोग करता हूं, जो प्रभावी रूप से सभी कोरटाइन को अपने परिणामों की सूची को पूरा करने और वापस करने की उम्मीद करता है।

आइए हम ध्यान दें कि कोरटाइन का उपयोग किसी भी एक बिंदु के साथ पुनरावृत्ति के लिए अच्छी तरह से मेल खाता है, जिस पर किसी भी संख्या में कॉरटॉयन्स मौजूद हैं, कॉल फ़ंक्शन के दौरान उनके अनुरोधों के जवाब की प्रतीक्षा कर रहा है और आई / ओ ऑपरेशन पूरा होने के बाद निष्पादन को फिर से शुरू कर रहा है। यह हमें एक सुरुचिपूर्ण और (आसानी से) पठनीय कॉरटिन में जटिल व्यवहार को व्यक्त करने की अनुमति देता है।

"बहुत आसान है," आप कहते हैं? ठीक है, चलो एक पायदान ऊपर चलते हैं।

गोली मार दी और भूल गया

कल्पना करें कि आप अपने आप को उन पोस्टों के साथ एक ई-मेल संदेश भेजना चाहते हैं जिनमें एक निश्चित मूल्य से अधिक टिप्पणियां हैं, और आप इसे उसी तरह से करना चाहते हैं जैसे हम पोस्ट ट्री के चारों ओर गए थे। हम इसे प्राप्त करने के लिए पुनरावर्ती फ़ंक्शन के अंत में केवल स्टेटमेंट जोड़ सकते हैं:

 async def post_number_of_comments(loop, session, post_id): """           . """ url = URL_TEMPLATE.format(post_id) response = await fetch(session, url) if 'kids' not in response: #  .   return 0 #       . number_of_comments = len(response['kids']) #       tasks = [post_number_of_comments(loop, session, kid_id) for kid_id in response['kids']] #      results = await asyncio.gather(*tasks) #            number_of_comments += sum(results) log.debug('{:^6} > {} comments'.format(post_id, number_of_comments)) #        if number_of_comments > MIN_COMMENTS: await log_post(response) return number_of_comments async def log_post(post): """   . """ await asyncio.sleep(random() * 3) log.info("Post logged") 

हां, मैंने asyncio.sleep का इस्तेमाल किया। यह आखिरी बार है। मैं वादा करता हूं।

 [09:41:02] Post logged [09:41:04] Post logged [09:41:06] Post logged [09:41:06] > Calculating comments took 6.35 seconds and 73 fetches [09:41:06] -- Post 8863 has 72 comments 

यह पहले की तुलना में काफी धीमा है!
कारण यह है कि, जैसा कि हमने पहले चर्चा की थी, भविष्य पूरा होने तक कोरटाइन निष्पादन का इंतजार करें, लेकिन चूंकि हमें लॉगिंग के परिणाम की आवश्यकता नहीं है, इसलिए ऐसा करने का कोई वास्तविक कारण नहीं है।

हमें अपने कॉरआउट के साथ "शूट और भूलने" की ज़रूरत है, और चूँकि हम वेट का उपयोग करके इसे खत्म करने के लिए इंतजार नहीं कर सकते, हमें बिना रुके कॉरआउट शुरू करने के लिए एक और रास्ता चाहिए। Asyncio एपीआई पर एक त्वरित नज़र सुनिश्चित करें_फिनाइल फ़ंक्शन मिलेगा जो कोरटाइन को चलाने के लिए शेड्यूल करेगा, इसे टास्क ऑब्जेक्ट में लपेटेगा , और इसे वापस लौटाएगा। यह याद करते हुए कि इससे पहले कि कोरटाइन की योजना बनाई गई थी, घटनाओं का एक चक्र भविष्य में किसी बिंदु पर हमारे कोरआउट के परिणाम को नियंत्रित करेगा, जब एक और कोरआउट उम्मीद की स्थिति में होगा।

महान, आइए निम्न के रूप में log_post की प्रतीक्षा करें:

 async def post_number_of_comments(loop, session, post_id): """           . """ url = URL_TEMPLATE.format(post_id) response = await fetch(session, url) if 'kids' not in response: # base case, there are no comments return 0 #       . number_of_comments = len(response['kids']) #       tasks = [post_number_of_comments( loop, session, kid_id) for kid_id in response['kids']] #      results = await asyncio.gather(*tasks) #            number_of_comments += sum(results) log.debug('{:^6} > {} comments'.format(post_id, number_of_comments)) #        if number_of_comments > MIN_COMMENTS: asyncio.ensure_future(log_post(response)) return number_of_comments 

 [09:42:57] > Calculating comments took 1.69 seconds and 73 fetches [09:42:57] -- Post 8863 has 72 comments [09:42:57] Task was destroyed but it is pending! task: <Task pending coro=<log_post() done, defined at 02_fire_and_forget.py:82> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x1109197f8>()]>> [09:42:57] Task was destroyed but it is pending! task: <Task pending coro=<log_post() done, defined at 02_fire_and_forget.py:82> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x110919948>()]>> [09:42:57] Task was destroyed but it is pending! task: <Task pending coro=<log_post() done, defined at 02_fire_and_forget.py:82> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x110919978>()]>> 

अहम, भयानक कार्य नष्ट हो गया था लेकिन यह लंबित है! दुनिया भर में asyncio उपयोगकर्ताओं की सता। अच्छी खबर यह है कि हम उस समय से वापस आ गए हैं जो हमने पहले प्राप्त किया था (1.69 पी।), बुरी खबर यह है कि एसिंकोयो को शॉट-एंड-भूल श्रेणी से परे जाना पसंद नहीं है।

समस्या यह है कि हम अपने कार्य log_post समय को पूरा किए बिना पोस्ट_number_of_comments coroutine का परिणाम प्राप्त करने के बाद जबरदस्ती इवेंट लूप को बंद कर देते हैं।

हमारे पास दो विकल्प हैं:
हम या तो इवेंट लूप को रन_ अफसर का उपयोग करते हुए पूरी तरह से काम करते हैं और स्क्रिप्ट को मैन्युअल रूप से समाप्त करते हैं, या हम सभी काम करने वाले कार्यों को खोजने के लिए टास्क क्लास की ऑल_ट्यूस पद्धति का उपयोग करते हैं और टिप्पणियों की संख्या समाप्त होने तक प्रतीक्षा करते हैं।

चलिए पोस्ट_नंबर_ऑफ_कॉम्मेंट्स के हमारे कॉल के बाद परिवर्तन करके इस स्थिति से जल्दी से बाहर निकलने की कोशिश करते हैं:

 if __name__ == '__main__': args = parser.parse_args() if args.verbose: log.setLevel(logging.DEBUG) post_id = id_from_HN_url(args.url) if args.url else args.id loop = asyncio.get_event_loop() with aiohttp.ClientSession(loop=loop) as session: now = datetime.now() comments = loop.run_until_complete( post_number_of_comments(loop, session, post_id)) log.info( '> Calculating comments took {:.2f} seconds and {} fetches'.format( (datetime.now() - now).total_seconds(), fetch_counter)) log.info("-- Post {} has {} comments".format(post_id, comments)) pending_tasks = [ task for task in asyncio.Task.all_tasks() if not task.done()] loop.run_until_complete(asyncio.gather(*pending_tasks)) loop.close() 

 [09:47:29] > Calculating comments took 1.72 seconds and 73 fetches [09:47:29] — Post 8863 has 72 comments [09:47:30] Post logged [09:47:31] Post logged [09:47:32] Post logged 

अब हमें यकीन है कि लॉगिंग कार्य पूर्ण हो चुके हैं।
यह धारणा कि ऑल_टेस विधि उन मामलों में ठीक काम करती है, जिनसे हम निपट रहे हैं, यह एक बढ़िया विचार है जब हमारे ईवेंट लूप में उचित रूप से कार्य किए जाते हैं, लेकिन अधिक जटिल मामलों में किसी भी कार्य को निष्पादित किया जा सकता है, जिसका स्रोत हमारे कोड के बाहर स्थित हो सकता है। ।

एक अन्य तरीका यह है कि हम स्वतंत्र रूप से पंजीकृत किए गए सभी कारआउट्स को स्वतंत्र रूप से पंजीकृत करने और निष्पादित करने की अनुमति देने के बाद आदेश को बहाल करें, जिसे पहले स्थगित कर दिया गया था,
जैसे ही टिप्पणियों की गिनती पूरी होती है। जैसा कि आप जानते हैं, ens_future फ़ंक्शन एक टास्क ऑब्जेक्ट देता है। हम अपने कार्यों को कम प्राथमिकता के साथ पंजीकृत करने के लिए इसका उपयोग कर सकते हैं। आइए बस एक कार्य_सूची सूची को परिभाषित करें और इसमें भविष्य को संचित करें:

 async def post_number_of_comments(loop, session, post_id): """Retrieve data for current post and recursively for all comments. """ url = URL_TEMPLATE.format(post_id) response = await fetch(session, url) if 'kids' not in response: # base case, there are no comments return 0 # calculate this post's comments as number of comments number_of_comments = len(response['kids']) # create recursive tasks for all comments tasks = [post_number_of_comments( loop, session, kid_id) for kid_id in response['kids']] # schedule the tasks and retrieve results results = await asyncio.gather(*tasks) # reduce the descendents comments and add it to this post's number_of_comments += sum(results) log.debug('{:^6} > {} comments'.format(post_id, number_of_comments)) # Log if number of comments is over a threshold if number_of_comments > MIN_COMMENTS: # Add the future to the registry task_registry.append(asyncio.ensure_future(log_post(response))) return number_of_comments # (... ommitted code ...) # if __name__ == '__main__': args = parser.parse_args() if args.verbose: log.setLevel(logging.DEBUG) post_id = id_from_HN_url(args.url) if args.url else args.id loop = asyncio.get_event_loop() task_registry = [] # define our task registry with aiohttp.ClientSession(loop=loop) as session: now = datetime.now() comments = loop.run_until_complete( post_number_of_comments(loop, session, post_id)) log.info( '> Calculating comments took {:.2f} seconds and {} fetches'.format( (datetime.now() - now).total_seconds(), fetch_counter)) log.info("-- Post {} has {} comments".format(post_id, comments)) pending_tasks = [task for task in task_registry if not task.done()] loop.run_until_complete(asyncio.gather(*pending_tasks)) loop.close() 

 [09:53:46] > Calculating comments took 1.68 seconds and 73 fetches [09:53:46] — Post 8863 has 72 comments [09:53:46] Post logged [09:53:48] Post logged [09:53:49] Post logged 

हम निम्नलिखित सबक सीखते हैं - एसिंको को अजवाइन की तरह वितरित कार्य कतार के रूप में नहीं देखा जाना चाहिए। सभी कार्यों को एक धागे में लॉन्च किया जाता है और घटनाओं के चक्र को तदनुसार प्रबंधित किया जाना चाहिए, जिससे आप कार्यों को पूरा करने के लिए समय आवंटित कर सकते हैं।

जो अन्य आम तौर पर स्वीकृत पैटर्न की ओर जाता है:

समय-समय पर ट्रिगर किए गए कोरटाइन

एचएन के बारे में हमारे उदाहरण के साथ जारी है (और हमने पहले एक बहुत अच्छा काम किया था), हमने फैसला किया
जो उपलब्ध होते ही HN प्रकाशन पर टिप्पणियों की संख्या की गणना करने के लिए निर्णायक रूप से महत्वपूर्ण है और वे हाल ही में 5 प्रविष्टियों की सूची में हैं।

एचएन एपीआई पर एक त्वरित नज़र एक समापन बिंदु दिखाती है जो पिछले 500 रिकॉर्ड लौटाता है। महान, इसलिए हम नए प्रकाशनों को प्राप्त करने और उन पर टिप्पणियों की संख्या की गणना करने के लिए बस इस समापन बिंदु को मतदान कर सकते हैं, हर पांच सेकंड में कह सकते हैं।

खैर, चूंकि अब हम समय-समय पर मतदान के लिए आगे बढ़ रहे हैं, हम बस लूप करते समय अनंत का उपयोग कर सकते हैं, मतदान कार्य को पूरा करने के लिए प्रतीक्षा करें ( प्रतीक्षा करें), और आवश्यक समय के लिए सो जाएं (कॉल स्लीप )। मैंने डायरेक्ट पोस्ट URL का उपयोग करने के बजाय शीर्ष प्रविष्टियाँ प्राप्त करने के लिए कुछ छोटे बदलाव किए।

 """ An example of periodically scheduling coroutines using an infinite loop of awaiting and sleeping. """ import asyncio import argparse import logging from datetime import datetime import aiohttp import async_timeout LOGGER_FORMAT = '%(asctime)s %(message)s' URL_TEMPLATE = "https://hacker-news.firebaseio.com/v0/item/{}.json" TOP_STORIES_URL = "https://hacker-news.firebaseio.com/v0/topstories.json" FETCH_TIMEOUT = 10 parser = argparse.ArgumentParser(description='Calculate the number of comments of the top stories in HN.') parser.add_argument('--period', type=int, default=5, help='Number of seconds between poll') parser.add_argument('--limit', type=int, default=5,help='Number of new stories to calculate comments for') parser.add_argument('--verbose', action='store_true', help='Detailed output') logging.basicConfig(format=LOGGER_FORMAT, datefmt='[%H:%M:%S]') log = logging.getLogger() log.setLevel(logging.INFO) fetch_counter = 0 async def fetch(session, url): """  URL   aiohttp,    JSON .     aiohttp    . """ global fetch_counter with async_timeout.timeout(FETCH_TIMEOUT): fetch_counter += 1 async with session.get(url) as response: return await response.json() async def post_number_of_comments(loop, session, post_id): """          . """ url = URL_TEMPLATE.format(post_id) response = await fetch(session, url) if 'kids' not in response: # base case, there are no comments return 0 #     ,    (  ) number_of_comments = len(response['kids']) #       tasks = [post_number_of_comments( loop, session, kid_id) for kid_id in response['kids']] #      results = await asyncio.gather(*tasks) #            number_of_comments += sum(results) log.debug('{:^6} > {} comments'.format(post_id, number_of_comments)) return number_of_comments async def get_comments_of_top_stories(loop, session, limit, iteration): """     HN. """ response = await fetch(session, TOP_STORIES_URL) tasks = [post_number_of_comments(loop, session, post_id) for post_id in response[:limit]] results = await asyncio.gather(*tasks) for post_id, num_comments in zip(response[:limit], results): log.info("Post {} has {} comments ({})".format(post_id, num_comments, iteration)) async def poll_top_stories_for_comments(loop, session, period, limit): """         . """ global fetch_counter iteration = 1 while True: now = datetime.now() log.info("Calculating comments for top {} stories. ({})".format(limit, iteration)) await get_comments_of_top_stories(loop, session, limit, iteration) log.info('> Calculating comments took {:.2f} seconds and {} fetches'.format( (datetime.now() - now).total_seconds(), fetch_counter)) log.info("Waiting for {} seconds...".format(period)) iteration += 1 fetch_counter = 0 await asyncio.sleep(period) if __name__ == '__main__': args = parser.parse_args() if args.verbose: log.setLevel(logging.DEBUG) loop = asyncio.get_event_loop() with aiohttp.ClientSession(loop=loop) as session: loop.run_until_complete( poll_top_stories_for_comments( loop, session, args.period, args.limit)) loop.close() 

 [10:14:03] Calculating comments for top 5 stories. (1) [10:14:06] Post 13848196 has 31 comments (1) [10:14:06] Post 13849430 has 37 comments (1) [10:14:06] Post 13849037 has 15 comments (1) [10:14:06] Post 13845337 has 128 comments (1) [10:14:06] Post 13847465 has 27 comments (1) [10:14:06] > Calculating comments took 2.96 seconds and 244 fetches [10:14:06] Waiting for 5 seconds… [10:14:11] Calculating comments for top 5 stories. (2) [10:14:14] Post 13848196 has 31 comments (2) [10:14:14] Post 13849430 has 37 comments (2) [10:14:14] Post 13849037 has 15 comments (2) [10:14:14] Post 13845337 has 128 comments (2) [10:14:14] Post 13847465 has 27 comments (2) [10:14:14] > Calculating comments took 3.04 seconds and 244 fetches [10:14:14] Waiting for 5 seconds… 

महान, लेकिन एक छोटी सी समस्या है: यदि आपने समय टिकट पर ध्यान दिया,
तब कार्य हर 5 सेकंड में सख्ती से शुरू नहीं होता है, यह get_comments_of_top_stories के पूरा होने के 5 सेकंड बाद शुरू होता है। जब तक हम अपने परिणाम वापस नहीं लेते तब तक प्रतीक्षा और अवरोधन के उपयोग के परिणाम।
जब यह कार्य पांच सेकंड से अधिक समय लेता है, तो ये सुविधाएँ समस्या उत्पन्न नहीं करती हैं। इसके अलावा, जब coroutine को अनंत बनाने के लिए डिज़ाइन किया गया हो, तो _run_until का उपयोग करना गलत लगता है।

अच्छी खबर यह है कि अब हम ens_future पर विशेषज्ञ हैं, और हम केवल प्रतीक्षा का उपयोग करने के बजाय इसे कोड में धकेल सकते हैं ...

 async def poll_top_stories_for_comments(loop, session, period, limit): """         . """ global fetch_counter iteration = 1 while True: now = datetime.now() log.info("Calculating comments for top {} stories. ({})".format( limit, iteration)) asyncio.ensure_future( get_comments_of_top_stories(loop, session, limit, iteration)) log.info( '> Calculating comments took {:.2f} seconds and {} fetches'.format( (datetime.now() - now).total_seconds(), fetch_counter)) log.info("Waiting for {} seconds...".format(period)) iteration += 1 fetch_counter = 0 await asyncio.sleep(period) 

  [10:55:40] Calculating comments for top 5 stories. (1) [10:55:40] > Calculating comments took 0.00 seconds and 0 fetches [10:55:40] Waiting for 5 seconds… [10:55:43] Post 13848196 has 32 comments (1) [10:55:43] Post 13849430 has 48 comments (1) [10:55:43] Post 13849037 has 16 comments (1) [10:55:43] Post 13845337 has 129 comments (1) [10:55:43] Post 13847465 has 29 comments (1) [10:55:45] Calculating comments for top 5 stories. (2) [10:55:45] > Calculating comments took 0.00 seconds and 260 fetches [10:55:45] Waiting for 5 seconds… [10:55:48] Post 13848196 has 32 comments (2) [10:55:48] Post 13849430 has 48 comments (2) [10:55:48] Post 13849037 has 16 comments (2) [10:55:48] Post 13845337 has 129 comments (2) [10:55:48] Post 13847465 has 29 comments (2) 

अहम ... ठीक है, अच्छी खबर यह है कि टाइमस्टैम्प ठीक पांच सेकंड बाद स्थित है, लेकिन 0.00 सेकंड क्या है और कोई नमूने नहीं हैं और फिर अगला पुनरावृत्ति शून्य सेकंड और 260 नमूने लेता है?

यह प्रतीक्षा से बचने के परिणामों में से एक है, अब हम कोरटाइन को ब्लॉक नहीं करते हैं और बस अगली पंक्ति पर जाते हैं, जो शून्य सेकंड प्रिंट करता है और, पहली बार शून्य संदेशों को पुनर्प्राप्त करता है। ये काफी छोटे कार्य हैं, क्योंकि हम संदेशों के बिना रह सकते हैं, लेकिन क्या होगा अगर हमें कार्यों के परिणाम की आवश्यकता है?

फिर, मेरे दोस्त, हमें सहारा लेना होगा ... कॉलबैक (क्रिंगिंग (())

मुझे पता है, मुझे पता है, कॉलआउट से बचने के लिए कॉरटाइन का पूरा बिंदु है, लेकिन ऐसा इसलिए है क्योंकि लेख का नाटकीय उपशीर्षक "बाहर का इंतजार" है। अब हम प्रतीक्षित क्षेत्र में नहीं हैं, हमारे पास कार्यों के मैनुअल लॉन्च के साथ रोमांच है, जो हमारे उपयोग के मामले को आगे बढ़ाता है। यह आपको क्या देता है? स्पॉइलर

जैसा कि हमने पहले चर्चा की है, सुनिश्चित करें कि भविष्य में ऑब्जेक्ट _future देता है जिसमें हम _add_done कॉलबैक का उपयोग करके कॉलबैक जोड़ सकते हैं।

इससे पहले कि हम ऐसा करते हैं, और सही भ्रूण की गिनती करने के लिए, हम इस बिंदु पर आते हैं कि हमें URLFetcher वर्ग में अपने निष्कर्षण कोआउटीन को एनकैप्सुलेट करना होगा । इस मामले में, हम प्रत्येक कार्य के लिए एक उदाहरण बनाते हैं ताकि हमारे पास नमूनों की सही गिनती हो। वैसे भी बग को पेश करने वाले वैश्विक वैरिएबल को हम हटा देते हैं:

 """               ensure_future   sleep.      future,    ensure_future,         ,    URLFetcher   . """ import asyncio import argparse import logging from datetime import datetime import aiohttp import async_timeout LOGGER_FORMAT = '%(asctime)s %(message)s' URL_TEMPLATE = "https://hacker-news.firebaseio.com/v0/item/{}.json" TOP_STORIES_URL = "https://hacker-news.firebaseio.com/v0/topstories.json" FETCH_TIMEOUT = 10 parser = argparse.ArgumentParser(description='Calculate the number of comments of the top stories in HN.') parser.add_argument('--period', type=int, default=5, help='Number of seconds between poll') parser.add_argument('--limit', type=int, default=5, help='Number of new stories to calculate comments for') parser.add_argument('--verbose', action='store_true', help='Detailed output') logging.basicConfig(format=LOGGER_FORMAT, datefmt='[%H:%M:%S]') log = logging.getLogger() log.setLevel(logging.INFO) class URLFetcher(): """   URL     """ def __init__(self): self.fetch_counter = 0 async def fetch(self, session, url): """  URL   aiohttp,    JSON .     aiohttp    . """ with async_timeout.timeout(FETCH_TIMEOUT): self.fetch_counter += 1 async with session.get(url) as response: return await response.json() async def post_number_of_comments(loop, session, fetcher, post_id): """          . """ url = URL_TEMPLATE.format(post_id) response = await fetcher.fetch(session, url) #  .  . if response is None or 'kids' not in response: return 0 #     ,    (  ) number_of_comments = len(response['kids']) #       tasks = [post_number_of_comments( loop, session, fetcher, kid_id) for kid_id in response['kids']] # s     results = await asyncio.gather(*tasks) #            number_of_comments += sum(results) log.debug('{:^6} > {} comments'.format(post_id, number_of_comments)) return number_of_comments async def get_comments_of_top_stories(loop, session, limit, iteration): """    HN. """ fetcher = URLFetcher() # create a new fetcher for this task response = await fetcher.fetch(session, TOP_STORIES_URL) tasks = [post_number_of_comments( loop, session, fetcher, post_id) for post_id in response[:limit]] results = await asyncio.gather(*tasks) for post_id, num_comments in zip(response[:limit], results): log.info("Post {} has {} comments ({})".format( post_id, num_comments, iteration)) return fetcher.fetch_counter # return the fetch count async def poll_top_stories_for_comments(loop, session, period, limit): """         . """ iteration = 1 while True: log.info("Calculating comments for top {} stories. ({})".format(limit, iteration)) future = asyncio.ensure_future(get_comments_of_top_stories(loop, session, limit, iteration)) now = datetime.now() def callback(fut): fetch_count = fut.result() log.info('> Calculating comments took {:.2f} seconds and {} fetches'.format( (datetime.now() - now).total_seconds(), fetch_count)) future.add_done_callback(callback) log.info("Waiting for {} seconds...".format(period)) iteration += 1 await asyncio.sleep(period) if __name__ == '__main__': args = parser.parse_args() if args.verbose: log.setLevel(logging.DEBUG) loop = asyncio.get_event_loop() with aiohttp.ClientSession(loop=loop) as session: loop.run_until_complete( poll_top_stories_for_comments( loop, session, args.period, args.limit)) loop.close() 

  [12:23:40] Calculating comments for top 5 stories. (1) [12:23:40] Waiting for 5 seconds... [12:23:43] Post 13848196 has 38 comments (1) [12:23:43] Post 13849430 has 72 comments (1) [12:23:43] Post 13849037 has 19 comments (1) [12:23:43] Post 13848283 has 64 comments (1) [12:23:43] Post 13847465 has 34 comments (1) [12:23:43] > Calculating comments took 3.17 seconds and 233 fetches [12:23:45] Calculating comments for top 5 stories. (2) [12:23:45] Waiting for 5 seconds... [12:23:47] Post 13848196 has 38 comments (2) [12:23:47] Post 13849430 has 72 comments (2) [12:23:47] Post 13849037 has 19 comments (2) [12:23:47] Post 13848283 has 64 comments (2) [12:23:47] Post 13847465 has 34 comments (2) [12:23:47] > Calculating comments took 2.47 seconds and 233 fetches [12:23:50] Calculating comments for top 5 stories. (3) [12:23:50] Waiting for 5 seconds... 

ठीक है, पहले से ही बेहतर है, लेकिन चलो कॉलबैक अनुभाग पर ध्यान दें:

 async def poll_top_stories_for_comments(loop, session, period, limit): """         . """ iteration = 1 while True: log.info("Calculating comments for top {} stories. ({})".format( limit, iteration)) future = asyncio.ensure_future( get_comments_of_top_stories(loop, session, limit, iteration)) now = datetime.now() def callback(fut): fetch_count = fut.result() log.info( '> Calculating comments took {:.2f} seconds and {} fetches'.format( (datetime.now() - now).total_seconds(), fetch_count)) future.add_done_callback(callback) log.info("Waiting for {} seconds...".format(period)) iteration += 1 await asyncio.sleep(period) 

ध्यान दें कि कॉलबैक फ़ंक्शन को एक तर्क लेना होगा, जिसमें भविष्य की वस्तु स्वयं पारित हो जाएगी। हम URLFetcher इंस्टेंस से _get_comments_of_top कहानियों के परिणाम के रूप में प्राप्त करने की संख्या भी लौटाते हैं और भविष्य के परिणाम के रूप में यह डेटा प्राप्त करते हैं।

आप देखते हैं?मैंने तुमसे कहा था कि यह अच्छा होगा, लेकिन यहाँ निश्चित रूप से प्रतीक्षा नहीं है ।

callback-, API asyncio AbstractBaseLoop _call later _call at ,
- . , , poll_top_stories_for_comments :

 def poll_top_stories_for_comments(loop, session, period, limit, iteration=0): """     get_comments_of_top_stories. """ log.info("Calculating comments for top {} stories ({})".format( limit, iteration)) future = asyncio.ensure_future( get_comments_of_top_stories(loop, session, limit, iteration)) now = datetime.now() def callback(fut): fetch_count = fut.result() log.info( '> Calculating comments took {:.2f} seconds and {} fetches'.format( (datetime.now() - now).total_seconds(), fetch_count)) future.add_done_callback(callback) log.info("Waiting for {} seconds...".format(period)) iteration += 1 loop.call_later( period, partial( # or call_at(loop.time() + period) poll_top_stories_for_comments, loop, session, period, limit, iteration ) ) if __name__ == '__main__': args = parser.parse_args() if args.verbose: log.setLevel(logging.DEBUG) loop = asyncio.get_event_loop() with aiohttp.ClientSession(loop=loop) as session: poll_top_stories_for_comments( loop, session, args.period, args.limit) loop.run_forever() loop.close() 

. :

• , ensure_future .
  ( : )
• _poll_top_stories_for comments , , _run forever , .

, , , — ? ? URL:

 MAXIMUM_FETCHES = 5 class URLFetcher(): """   URL     """ def __init__(self): self.fetch_counter = 0 async def fetch(self, session, url): """  URL   aiohttp,    JSON .     aiohttp    . """ with async_timeout.timeout(FETCH_TIMEOUT): self.fetch_counter += 1 if self.fetch_counter > MAXIMUM_FETCHES: raise Exception('BOOM!') async with session.get(url) as response: return await response.json() 

  [12:51:00] Calculating comments for top 5 stories. (1) [12:51:00] Waiting for 5 seconds… [12:51:01] Exception in callback poll_top_stories_for_comments.<locals>.callback(<Task finishe…ion('BOOM!',)>) at 05_periodic_coroutines.py:121 handle: <Handle poll_top_stories_for_comments.<locals>.callback(<Task finishe…ion('BOOM!',)>) at 05_periodic_coroutines.py:121> Traceback (most recent call last): File “/Users/yeray/.pyenv/versions/3.6.0/lib/python3.6/asyncio/events.py”, line 126, in _run self._callback(*self._args) File “05_periodic_coroutines.py”, line 122, in callback fetch_count = fut.result() File “05_periodic_coroutines.py”, line 100, in get_comments_of_top_stories results = await asyncio.gather(*tasks) File “05_periodic_coroutines.py”, line 69, in post_number_of_comments response = await fetcher.fetch(session, url) File “05_periodic_coroutines.py”, line 58, in fetch raise Exception('BOOM!') Exception: BOOM! [12:51:05] Calculating comments for top 5 stories. (2) [12:51:05] Waiting for 5 seconds… 

, ?

क्या करें? , , : asyncio :