Как я могу разбить этот многопоточный скрипт python на «куски»?

Question

Я обрабатываю доменные имена 100k в CSV на основе результатов, полученных с Siteadvisor, используя urllib (не лучший метод, я знаю). Однако мой текущий скрипт создает слишком много потоков, и Python попадает в ошибки. Есть ли способ, которым я могу «разбить» этот скрипт для того, чтобы делать X количество доменов за раз (например, 10-20), чтобы предотвратить эти ошибки? Заранее спасибо.

import threading 
import urllib 

class Resolver(threading.Thread): 
    def __init__(self, address, result_dict): 
     threading.Thread.__init__(self) 
     self.address = address 
     self.result_dict = result_dict 

    def run(self): 
     try: 
      content = urllib.urlopen("http://www.siteadvisor.com/sites/" + self.address).read(12000) 
      search1 = content.find("didn't find any significant problems.") 
      search2 = content.find('yellow') 
      search3 = content.find('web reputation analysis found potential security') 
      search4 = content.find("don't have the results yet.") 

      if search1 != -1: 
       result = "safe" 
      elif search2 != -1: 
       result = "caution" 
      elif search3 != -1: 
       result = "warning" 
      elif search4 != -1: 
       result = "unknown" 
      else: 
       result = "" 

      self.result_dict[self.address] = result 

     except: 
      pass 


def main(): 
    infile = open("domainslist", "r") 
    intext = infile.readlines() 
    threads = [] 
    results = {} 
    for address in [address.strip() for address in intext if address.strip()]: 
     resolver_thread = Resolver(address, results) 
     threads.append(resolver_thread) 
     resolver_thread.start() 

    for thread in threads: 
     thread.join() 

    outfile = open('final.csv', 'w') 
    outfile.write("\n".join("%s,%s" % (address, ip) for address, ip in results.iteritems())) 
    outfile.close() 

if __name__ == '__main__': 
    main() 

Edit: новая версия, на основе предложений andyortlieb в.

import threading 
import urllib 
import time 

class Resolver(threading.Thread): 
    def __init__(self, address, result_dict, threads): 
     threading.Thread.__init__(self) 
     self.address = address 
     self.result_dict = result_dict 
     self.threads = threads 
    def run(self): 
     try: 
      content = urllib.urlopen("http://www.siteadvisor.com/sites/" + self.address).read(12000) 
      search1 = content.find("didn't find any significant problems.") 
      search2 = content.find('yellow') 
      search3 = content.find('web reputation analysis found potential security') 
      search4 = content.find("don't have the results yet.") 

      if search1 != -1: 
       result = "safe" 
      elif search2 != -1: 
       result = "caution" 
      elif search3 != -1: 
       result = "warning" 
      elif search4 != -1: 
       result = "unknown" 
      else: 
       result = "" 

      self.result_dict[self.address] = result 

      outfile = open('final.csv', 'a') 
      outfile.write(self.address + "," + result + "\n") 
      outfile.close() 
      print self.address + result 

      threads.remove(self) 
     except: 
      pass 


def main(): 
    infile = open("domainslist", "r") 
    intext = infile.readlines() 
    threads = [] 
    results = {} 

    for address in [address.strip() for address in intext if address.strip()]: 
     loop=True 
     while loop: 
      if len(threads) < 20: 
       resolver_thread = Resolver(address, results, threads) 
       threads.append(resolver_thread) 
       resolver_thread.start() 
       loop=False 
      else: 
       time.sleep(.25) 


    for thread in threads: 
     thread.join() 

# removed so I can track the progress of the script 
# outfile = open('final.csv', 'w') 
# outfile.write("\n".join("%s,%s" % (address, ip) for address, ip in results.iteritems())) 
# outfile.close() 

if __name__ == '__main__': 
    main() 

Answer 1

Это может быть своего рода жестким, но вы можете передать темы к решающему, так что, когда резольвера .run завершен, он может называть threads.remove (self)

Тогда вы можете вложить некоторые условия, чтобы потоки создавались только в том случае, если есть место для них, и если нет места, они ждут, пока не появится ,

for address in [address.strip() for address in intext if address.strip()]: 
     loop=True 
     while loop: 
      if len(threads)<20: 
       resolver_thread = Resolver(address, results, threads) 
       threads.append(resolver_thread) 
       resolver_thread.start() 
       loop=False 
      else: 
       time.sleep(.25) 

Answer 2

Ваш существующий код будет работать красиво - просто измените метод __init__ внутри Resolver взять в дополнительном списке адресов вместо одного в то время, так что вместо того, чтобы один поток для каждого адреса, у вас есть один поток на каждые 10 (например). Таким образом, вы не будете перегружать потоки.

Вам, очевидно, придется слегка изменить run, чтобы он проходил через массив адресов, а не один self.address.

Я могу выполнить быстрый пример, если вы хотите, но из-за качества вашего кода мне кажется, что вы сможете справиться с ним довольно легко.

Надеюсь, это поможет!

EDIT Пример ниже в соответствии с запросом. Обратите внимание, что вам нужно будет изменить main, чтобы отправить список адресов адресов Resolver вместо одного адреса - я не мог справиться с этим для вас, не зная больше о формате вашего файла и способах сохранения адресов. Примечание - вы могли бы сделать метод run с вспомогательной функцией, но я думал, что это могло бы быть более понятным, как пример

class Resolver(threading.Thread): 
    def __init__(self, addresses, result_dict): 
     threading.Thread.__init__(self) 
     self.addresses = addresses # Now takes in a list of multiple addresses 
     self.result_dict = result_dict 

    def run(self): 
     for address in self.addresses: # do your existing code for every address in the list 
      try: 
       content = urllib.urlopen("http://www.siteadvisor.com/sites/" + address).read(12000) 
       search1 = content.find("didn't find any significant problems.") 
       search2 = content.find('yellow') 
       search3 = content.find('web reputation analysis found potential security') 
       search4 = content.find("don't have the results yet.") 

       if search1 != -1: 
        result = "safe" 
       elif search2 != -1: 
        result = "caution" 
       elif search3 != -1: 
        result = "warning" 
       elif search4 != -1: 
        result = "unknown" 
       else: 
        result = "" 

       self.result_dict[address] = result 
      except: 
       pass