在Python中维护一个可自动清除的线程列表

Question

我维护一个threads列表，并且希望在线程完成后从列表中自动删除它们。

我发现了这个方法：

import threading, time

def f(seconds, info):
    print('starting', seconds)
    time.sleep(seconds)
    print('finished', seconds)
    threads.remove(info['thread'])

def newaction(seconds):
    info = {}
    thread = threading.Thread(target=f, args=(seconds, info))
    info['thread'] = thread
    thread.start()
    threads.append(thread)

threads = []
newaction(1)
newaction(2)
for _ in range(10):
    time.sleep(0.3)
    print(threads)

它的作用是：

starting 1
starting 2
[<Thread(Thread-1, started 1612)>, <Thread(Thread-2, started 712)>]
[<Thread(Thread-1, started 1612)>, <Thread(Thread-2, started 712)>]
[<Thread(Thread-1, started 1612)>, <Thread(Thread-2, started 712)>]
finished 1
[<Thread(Thread-2, started 712)>]
[<Thread(Thread-2, started 712)>]
[<Thread(Thread-2, started 712)>]
finished 2
[]
[]
[]
[]

但是，必须通过一个迪克info的事实是有点麻烦。我用它是因为很明显我不能通过thread在args.

thread = threading.Thread(target=f, args=(seconds, thread))  
#                                                     ^ not created yet!

...when Thread对象尚未创建！

中是否有更自然的方法来维护线程的自动清除列表？

Answer 1

您有current_thread()函数。

import threading, time

def f(seconds):
    print('starting', seconds)
    time.sleep(seconds)
    print('finished', seconds)
    threads.remove(threading.current_thread())

def newaction(seconds):
    thread = threading.Thread(target=f, args=(seconds,))
    thread.start()
    threads.append(thread)

threads = []
newaction(1)
newaction(2)
for _ in range(10):
    time.sleep(0.3)
    print(threads)

输出：

starting 1
starting 2
[<Thread(Thread-1, started 4588)>, <Thread(Thread-2, started 4388)>]
[<Thread(Thread-1, started 4588)>, <Thread(Thread-2, started 4388)>]
[<Thread(Thread-1, started 4588)>, <Thread(Thread-2, started 4388)>]
finished 1
[<Thread(Thread-2, started 4388)>]
[<Thread(Thread-2, started 4388)>]
[<Thread(Thread-2, started 4388)>]
finished 2
[]
[]
[]
[]

Answer 2

子类线程将产生一个具有自然语法和安全的位置来保存线程列表的解决方案。您也不必在要在另一个线程中运行的每个函数的末尾包含一个删除线程的指令。只需使用子类即可。

import threading, time

class AutoRemovingThread(threading.Thread):
  threads = []
  def __init__(self, func, *args, **kwargs):
    super().__init__()
    self.threads.append(self)
    self.func = func
    self.args = args
    self.kwargs = kwargs
  def run(self):
    self.func(*self.args, **self.kwargs)
    self.threads.remove(self)

def f(seconds):
  print('starting', seconds)
  time.sleep(seconds)
  print('finished', seconds)

def newaction(seconds):
  AutoRemovingThread(f, seconds).start()

newaction(1)
newaction(2)
for _ in range(10):
  time.sleep(0.3)
  print(AutoRemovingThread.threads)

产出：

starting 1
starting 2
[<AutoRemovingThread(Thread-1, started 8436)>, <AutoRemovingThread(Thread-2, started 1072)>]
[<AutoRemovingThread(Thread-1, started 8436)>, <AutoRemovingThread(Thread-2, started 1072)>]
[<AutoRemovingThread(Thread-1, started 8436)>, <AutoRemovingThread(Thread-2, started 1072)>]
finished 1
[<AutoRemovingThread(Thread-2, started 1072)>]
[<AutoRemovingThread(Thread-2, started 1072)>]
[<AutoRemovingThread(Thread-2, started 1072)>]
finished 2
[]
[]
[]
[]

python-3.8

Answer 3

import threading

def get_status_of_threads():
    current_threads = threading.enumerate()

    thread_data = []

    for item in current_threads:
        try:
            print(str(item.target))
        except AttributeError:
            print("item", str(item))
        thread_data.append({"thread_name": item.getName(), "status": int(item.is_alive()), "id": item.ident})
    return thread_data

上面的代码是在Python2.7中测试的，如果您想要持续监视线程，可以在单独的线程中使用它，或者可以将其公开为API，这样您就可以随时检查。这也有助于减少同样的资源浪费。

对于API，您可以使用json2html.convert({"thread_data":thread_data})函数以更美观的方式以表格形式显示它。