稍后在 python 中杀死线程

Question

我有一个带有线程的 python 代码，如果在例如 1 小时内线程没有完成，我需要完成所有线程并完成脚本，如果这个小时没有完成，请等待我的所有线程完成。

我尝试使用守护线程，并使用一小时的睡眠，如果一小时已完成，请使用：sys.exit() 但它对我不起作用，因为总是等待我的睡眠线程，然后我的脚本等到线程完成，sys.exit() 不起作用。

import socket, threading, time, sys
from sys import argv
import os

acc_time=0
transactions_ps=5

ins = open(sys.argv[1],'r')
msisdn_list = []
for line in ins:
    msisdn_list.append (line.strip('\n'))
   # print line
ins.close()


def worker(msisdn_list):
    semaphore.acquire()
    global transactions_ps
    print "  *****  ", threading.currentThread().getName(), "Lanzado"
    count=1
    acc_time=0
    print "len: ",len(msisdn_list)
    for i in msisdn_list:
        try:
            init=time.time()
            time.sleep(2)
            print "sleeping...",i
            time.sleep(4)
            final=time.time()
            acc_time = acc_time+final-init
            print acc_time
        except IOError:
                print "Connection failed",sys.exc_info()[0]

    print "Deteniendo ",threading.currentThread().getName()
    semaphore.release()
def kill_process(secs_to_die):
    time.sleep(secs_to_die)
    sys.exit()

seconds_to_die=3600

thread_kill = threading.Thread(target = kill_process, args=(seconds_to_die,))
thread_kill.start()

max_con=5
semaphore = threading.BoundedSemaphore(max_con)
for i in range(0,28,transactions_ps):
    w = threading.Thread(target=worker, args=(msisdn_list[i:i+transactions_ps-1],))
    w.setDaemon(True)
    w.start()

怎么办

Answer 1

可以解决问题的代码的最小更改是threading.Barrier：

barrier = Barrier(number_of_threads, timeout=3600)
# create (number_of_threads - 1) threads, pass them barrier
# each thread calls barrier.wait() on exit
barrier.wait() # after number_of_threads .wait() calls or on timeout it returns

更简单的替代方法是使用创建守护线程的multiprocessing.dummy.Pool：

from multiprocessing.dummy import Pool # use threads

start = timer()
endtime = start + 3600
for result in pool.imap_unordered(work, args):
    if timer() > endtime:
       exit("timeout") 

代码在工作项完成之前不会超时，即，它预计处理列表中的单个项目不会花费很长时间。

完整示例：

#!/usr/bin/env python3
import logging
import multiprocessing as mp
from multiprocessing.dummy import Pool
from time import monotonic as timer, sleep

info = mp.get_logger().info

def work(i):
    info("start %d", i)
    sleep(1)
    info("end %d", i)

seconds_to_die = 3600
max_con = 5
mp.log_to_stderr().setLevel(logging.INFO) # enable logging
pool = Pool(max_con) # no more than max_con at a time
start = timer()
endtime = start + seconds_to_die
for _ in pool.imap_unordered(work, range(10000)):
    if timer() > endtime:
        exit("timeout")

Answer 2

你可以参考这个KThread的实现：

http://python.todaysummary.com/q_python_45717.html