再次执行失败的线程

Question

所以我有一个使用大约 50k 线程的脚本，但一次只运行 10 个。我为此使用线程库，并使用 BoundedSemaphore 将线程一次限制为 10 个。在某些情况下，所有线程都没有足够的内存，但重要的是所有线程都得到处理，所以我想重复那些因内存不足而被杀死的线程。

import some_other_script, threading


class myThread (threading.Thread):
    def __init__(self, item):
        threading.Thread.__init__(self)
        self.item = item
    def run(self):
        threadLimiter.acquire()
        some_other_script.method(self.item)
        somelist.remove(self.item)
        threadLimiter.release()


threadLimiter = threading.BoundedSemaphore(10)

somelist = ['50,000 Items','.....]
for item in somelist:
    myThread(item).start()

正如您所看到的，到目前为止我能想到的唯一想法是使用somelist.remove(self.item) 从每个线程内的列表中删除已处理的项目。 （每个项目都是唯一的，并且仅在列表中出现一次）。 我的想法是我可以在 for 循环周围运行一个 while 循环，以检查它是否仍然包含不起作用的项目，因为在 for 循环完成后线程没有完成，因此列表不为空。 我想要做的是捕捉那些失败的，因为系统内存不足并再次执行它们（如果需要，再次执行）。

非常感谢您！

Answer 1

这解决了过多的活动线程问题和您的问题：

    def get_items():
          threads = threading.enumerate()
          items = set()
          for thr in threads:
              if isinstance(thr, myThread): items.add(thr.item)
          return items
    def manageThreads(howmany):
         while bigset:
             items = get_items()
             items_to_add = bigset.difference(items)
             while len(items) < howmany:
                 item = items_to_add.pop()
                 processor = myThread(item)
                 processor.start()
             with thread_done:    
                 thread_done.wait()
   thread_done = threading.Condition()
   bigset = set(["50,000 items", "..."])
   manageThreads(10)

mythread类的run方法：

def run(self):
    try:
        some_other_script.method(self.item)
        bigset.remove(self.item)
    finally:
        with thread_done:
            thread_done.notify()

Threading.enumerate() 返回当前活动线程对象的列表。因此，manageThreads 函数最初创建 10 个线程，然后等待一个线程完成，然后再次检查线程数，依此类推。如果线程耗尽内存或在处理过程中发生其他错误，它不会从bigset 中删除该项目，从而导致它被管理器重新排队到另一个线程中。