首页
学习
活动
专区
圈层
工具
发布
社区首页 >问答首页 >Python3.4 concurrent.futures.Executor不允许控件暂停并继续线程

Python3.4 concurrent.futures.Executor不允许控件暂停并继续线程
EN

Stack Overflow用户
提问于 2014-07-25 19:17:58
回答 1查看 2.6K关注 0票数 2

我正在使用concurrent.future.ThredPoolExecutor来执行多线程,我正在执行几个http服务,我希望在服务器关闭时,对线程的控制暂停执行,启动服务器,然后继续执行。

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

因此,concurrent.futures.Executor.shutdown()将向执行器发出信号,表示在执行当前未决的期货时,它应该释放它正在使用的任何资源。

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

实际上,由于在concurren.future中找不到暂停和恢复,所以我正在调用pause ()方法。因此,作为另一种选择,在线程完成执行后,我将从列表中删除urls。这样我就可以传递剩余的列表并回忆相同的方法。

以下是代码:

代码语言:javascript
复制
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:\logs\serverStopLog.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:\inetpub\wwwroot")
trigegerFunc()
EN

回答 1

Stack Overflow用户

发布于 2014-07-25 20:01:17

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

向执行者发出信号,在执行当前悬而未决的期货时,它应该释放它正在使用的任何资源,执行

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

代码语言:javascript
复制
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:\logs\serverStopLog.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操作。

票数 1
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接:

https://stackoverflow.com/questions/24963016

复制
相关文章

相似问题

领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档