क्यूई के रूप में जल्द ही प्रगतिशील वस्तुओं को उठाकर

Question

मैं Queue का उपयोग करके वस्तुओं की सूची के माध्यम से प्रगतिशील रूप से काम करने की अनुमति देने के लिए ठोस कार्यान्वयन की तलाश में हूं।

विचार यह है कि मैं 20+ डेटाबेस गहन कार्यों की सूची के माध्यम से कार्यकर्ताओं की एक निश्चित संख्या का उपयोग करना चाहता हूं और परिणाम वापस कर सकता हूं। मैं पाइथन को पांच पहले आइटमों से शुरू करना चाहता हूं और जैसे ही यह एक कार्य के साथ किया जाता है, कतार में अगले कार्य पर शुरू होता है।

इस प्रकार मैं वर्तमान में Threading के बिना ऐसा कर रहा हूं।

for key, v in self.sources.iteritems(): 
    # Do Stuff 

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

Answer 1

आप इसे स्वयं लागू कर सकते हैं, लेकिन पाइथन 3 पहले से ही Executor थ्रेड प्रबंधन के लिए आधारित समाधान के साथ आता है, जिसे आप the backported version इंस्टॉल करके पाइथन 2.x में उपयोग कर सकते हैं।

आपका कोड तो देखो की तरह

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: 
    future_to_key = {} 
    for key, value in sources.items(): 
     future_to_idday[executor.submit(do_stuff, value)] = key 
    for future in concurrent.futures.as_completed(future_to_key): 
     key = future_to_key[future] 
     result = future.result() 
     # process result 

Answer 2

आप python3 का उपयोग कर रहे हैं, तो मैं समवर्ती वायदा मॉड्यूल की सिफारिश कर सकता है। यदि आप python3 का उपयोग नहीं कर रहे हैं और थ्रेड (बनाम प्रक्रियाओं) से जुड़े नहीं हैं तो आप multiprocessing.Pool (हालांकि यह कुछ चेतावनी के साथ आता है और मुझे पूल के साथ परेशानी है जो मेरे अनुप्रयोगों में ठीक से बंद नहीं हो रहा है)। यदि आपको थियम्स का उपयोग करना होगा, तो Python2 में, आप खुद को लेखन कोड समाप्त कर सकते हैं - 5 थ्रेड उपभोक्ता कार्यों को चलाते हैं और उपभोक्ताओं के लिए उन्हें खोजने और संसाधित करने के लिए केवल कतार में कॉल (फ़ंक्शन + तर्क) को धक्का देते हैं।

Answer 3

आप केवल stdlib का उपयोग कर इसे कर सकता है:

#!/usr/bin/env python 
from multiprocessing.dummy import Pool # use threads 

def db_task(key_value): 
    try: 
     key, value = key_value 
     # compute result.. 
     return result, None 
    except Exception as e: 
     return None, e 

def main(): 
    pool = Pool(5) 
    for result, error in pool.imap_unordered(db_task, sources.items()): 
     if error is None: 
      print(result) 

if __name__=="__main__": 
    main()