我正在使用concurrent.future.ThredPoolExecutor来执行多线程,我正在执行几个http服务,我希望在服务器关闭时,对线程的控制暂停执行,启动服务器,然后继续执行。
服务器停机的触发器是,我正在检查某个文件在特定位置是否可用,然后我将不得不暂停执行。
因此,concurrent.futures.Executor.shutdown()将向执行器发出信号,表示在执行当前未决的期货时,它应该释放它正在使用的任何资源。
但是,当我使用executor的when ()方法时,它不是立即关闭线程,而是在完成整个执行之后调用shutting ()。
实际上,由于在concurren.future中找不到暂停和恢复,所以我正在调用pause ()方法。因此,作为另一种选择,在线程完成执行后,我将从列表中删除urls。这样我就可以传递剩余的列表并回忆相同的方法。
以下是代码:
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()发布于 2014-07-25 20:01:17
不能取消或暂停/恢复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:\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操作。
https://stackoverflow.com/questions/24963016
复制相似问题