Python 3.4 concurrent.futures.Executor 不提供暂停和恢复线程的控件

Question

我正在使用concurrent.future. threadpoolexecutor进行多线程，我正在执行一些http服务，我想控制线程在服务器宕机时暂停执行，启动服务器然后恢复执行。

服务器关闭的触发器是，我正在检查文件在特定位置是否可用，然后我将不得不暂停执行。

所以concurrent.futures.Executor.shutdown()会通知executor，当当前挂起的futures完成执行时，它应该释放正在使用的资源。

但是当我使用executor的shutdown()方法时，它不会立即关闭线程，而是在完成整个执行后调用shutdown()。

事实上，我正在调用shutdown()方法，因为我无法在concurrent .future中找到暂停和恢复。因此，作为另一种选择，我从列表中删除的url一旦线程完成执行。这样我就可以传递剩余的列表并调用相同的方法。

代码如下:

import concurrent.futures
import urllib.request
import os.path
import datetime
import sys
import pathlib
from errno import ENOENT, EACCES, EPERM
import time
import threading
listOfFilesFromDirectory =  []
webroot = settings.configuration.WEBSERVER_WEBROOT
WEBSERVER_PORT = settings.configuration.WEBSERVER_PORT
shutdown = False
def class myclass:
#populating the list with the urls from a file
def triggerMethod(path):
    try:
        for line in open(path):
            listOfFilesFromDirectory.append(line)
    except IOError as err:
        if err.errno == ENOENT:
            #logging.critical("document.txt file is missing")
            print("document.txt file is missing")
        elif err.errno in (EACCES, EPERM):
            #logging.critical("You are not allowed to read document.txt")
            print("You are not allowed to read document.txt")
        else:
            raise   
# calling this method to stop the threads and restart after a sleep of 100 secs, as the list will always have the urls that were not executed.
def stopExecutor(executor):
    filePath = "C:logsserverStopLog.txt"
    while not shutdown:
        time.sleep(5)
        if os.path.isfile(filePath):
            executor.shutdown( )
            time.sleep(100)
            runRegressionInMultipleThreads( )
            break
def load_url(url, timeout):
    conn = urllib.request.urlopen('http://localhost:' + WEBSERVER_PORT + "/" + url, timeout = timeout)
    return conn.info()
def trigegerFunc( ):
    # We can use a with statement to ensure threads are cleaned up promptly
    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
        # Start the load operations and mark each future with its URL
        future_to_url = {executor.submit(load_url, url, 60): url for url in listOfFilesFromDirectory}
        t = threading.Thread(target=stopExecutor, args=(executor))
        t.start()
        for future in concurrent.futures.as_completed(future_to_url):
            url = future_to_url[future]
            try:
                data = future.result()
            except Exception as exc:
                print('%r generated an exception: %s' % (url, exc))
                listOfFilesFromDirectory.remove(url)
            else:
                if data:
                    if "200" in data:
                        listOfFilesFromDirectory.remove(url)
                    else:
                        listOfFilesFromDirectory.remove(url)
                else:
                    listOfFilesFromDirectory.remove(url)
        shutdown = True
        t.join()                

triggerMethod("C:inetpubwwwroot")
trigegerFunc()

Answer 1

在Python中不能取消或暂停/恢复线程。executor.shutdown()完全按照您引用文档时所说的那样做:

通知执行器应该释放它正在使用的任何资源当前挂起的期货执行完成时。

请注意，保留部分-执行器只会在所有当前执行的任务完成后关闭。要获得所需的控制，需要在单独的进程中运行urllib调用，如下所示(这是脚本的简化版本):

import time
import os.path
import threading
import urllib.request
import multiprocessing
import concurrent.futures
from multiprocessing import cpu_count
shutdown = False
should_cancel = False
def stopTasks():
    global should_cancel
    filePath = "C:logsserverStopLog.txt"
    while not shutdown:
        time.sleep(5)
        if os.path.isfile(filePath):
            should_cancel = True
            break
def _load_url(num, timeout, q):
    conn = urllib.request.urlopen('http://localhost:' + WEBSERVER_PORT + 
                                  "/" + url, timeout=timeout)
    q.put(conn.info())
def load_url(num, timeout):
    q = multiprocessing.Queue()
    p = multiprocessing.Process(target=_load_url, args=(num, timeout, q))
    p.start()
    while p.is_alive():
        time.sleep(.5)
        if should_cancel:
            p.terminate()  # This will actually kill the process, cancelling the operation
            break # You could return something here that indicates it was cancelled, too.
    else:
        # We'll only enter this if we didn't `break` above.
        out = q.get()
        p.join()
        return out
def triggerFunc():
    global shutdown
    with concurrent.futures.ThreadPoolExecutor(max_workers=cpu_count()) as executor:
        # Start the load operations and mark each future with its URL
        future_to_url = {executor.submit(load_url, url, 60):
                             url for url in listOfFilesFromDirectory}
        t = threading.Thread(target=stopTasks)
        t.start()
        for future in concurrent.futures.as_completed(future_to_url):
            info = future.result()
            print("done: {}".format(info))
            # other stuff you do
        shutdown = True
        t.join()
if __name__ == "__main__":
    triggerFunc()

因为我们可以通过发送SIGTERM来终止子进程，所以我们可以在urlopen操作仍在进行时真正取消它。