我想我找到了合适的解决方案：
我创建了自己的小型控制台应用程序并挂接到它的消息队列以捕获关闭事件。
我还没有对其进行太多测试，也不知道这是否是一个好的解决方案，但也许它对某人有帮助。
首先是我基于 tkinter 的简单控制台的代码。它以黑色显示 stdout，以红色显示 stderr：
# a simple console based on tkinter to display stdout and stderr
class SimpleConsole(object):
def __init__(self, name):
self.root = Tk()
self.root.title(name)
self.init_ui()
def init_ui(self):
self.text_box = Text(self.root, wrap='word', height = 11, width=50)
self.text_box.grid(column=0, row=0, columnspan = 2, sticky='NSWE', padx=5, pady=5)
self.text_box.tag_config('std', foreground="black")
self.text_box.tag_config('err', foreground="red")
self.text_box.pack(side=LEFT, fill=BOTH, expand = YES)
self.text_box.yview()
self.yscrollbar = Scrollbar(self.root, orient=VERTICAL, command=self.text_box.yview)
self.yscrollbar.pack(side=RIGHT, fill=Y)
self.text_box["yscrollcommand"] = self.yscrollbar.set
sys.stdout = SimpleConsole.StdRedirector(self.text_box, "std")
sys.stderr = SimpleConsole.StdRedirector(self.text_box, "err")
self.update()
class StdRedirector(object):
def __init__(self, text_widget, tag):
self.text_space = text_widget
self.tag = tag
def write(self, string):
self.text_space.insert('end', string, self.tag)
self.text_space.see('end')
def flush(self):
pass
def update(self):
self.root.update()
def get_window_handle(self):
return int(self.root.wm_frame(), 16)
然后我创建了一个类，它挂接到我的控制台的消息队列并管理关闭：
#class to handle a graceful shutdown by hooking into windows message queue
class GracefulShutdown:
def __init__(self, handle):
self.shutdown_requested = False
self._shutdown_functions = []
self.handle = handle
try:
if os.name == 'nt':
# Make a dictionary of message names to be used for printing below
self.msgdict = {}
for name in dir(win32con):
if name.startswith("WM_"):
value = getattr(win32con, name)
self.msgdict[value] = name
# Set the WndProc to our function
self.oldWndProc = win32gui.SetWindowLong(self.handle, win32con.GWL_WNDPROC, self.my_wnd_proc)
if self.oldWndProc == 0:
raise NameError("wndProc override failed!")
self.message_map = {win32con.WM_QUERYENDSESSION: self.hdl_query_end_session,
win32con.WM_ENDSESSION: self.hdl_end_session,
win32con.WM_QUIT: self.hdl_quit,
win32con.WM_DESTROY: self.hdl_destroy,
win32con.WM_CLOSE: self.hdl_close}
# pass a shutdown message to windows
retval = windll.user32.ShutdownBlockReasonCreate(self.handle,c_wchar_p("I'm still saving data!"))
if retval == 0:
raise NameError("shutdownBlockReasonCreate failed!")
except Exception as e:
logging.exception("something went wrong during win32 shutdown detection setup")
#catches all close signals and passes it to our own functions; all other signals are passed to the original function
def my_wnd_proc(self, hwnd, msg, w_param, l_param):
# Display what we've got.
logging.debug(self.msgdict.get(msg), msg, w_param, l_param)
# Restore the old WndProc. Notice the use of wxin32api
# instead of win32gui here. This is to avoid an error due to
# not passing a callable object.
if msg == win32con.WM_DESTROY:
win32api.SetWindowLong(self.handle,
win32con.GWL_WNDPROC,
self.oldWndProc)
#simplify function for calling
def call_window_proc_old():
return win32gui.CallWindowProc(self.oldWndProc, hwnd, msg, w_param, l_param)
#either call our handle functions or call the original wndProc
return self.message_map.get(msg, call_window_proc_old)()
def hdl_query_end_session(self):
logging.info("WM_QUERYENDSESSION received")
self.shutdown_requested = True
#we have to return 0 here to prevent the windows shutdown until our application is closed
return 0
def hdl_end_session(self):
logging.info("WM_ENDSESSION received")
self.exit_gracefully()
return 0
def hdl_quit(self):
logging.info("WM_QUIT received")
self.shutdown_requested = True
return 0
def hdl_destroy(self):
logging.info("WM_DESTROY received")
return 0
def hdl_close(self):
logging.info("WM_CLOSE received")
self.shutdown_requested = True
return 0
def exit_gracefully(self):
logging.info("shutdown request received")
self.shutdown_requested = True
for func in self._shutdown_functions:
try:
func()
except:
logging.exception("Exception during shutdown function:")
logging.info("shutdown request done, bye!")
exit(0)
def add_cleanup_function(self, function):
self._shutdown_functions.append(function)
这里有一些“主要”代码来启动这两个类并对其进行测试：
if __name__ == "__main__":
import time
from logging.handlers import RotatingFileHandler
#setup own console window
console = SimpleConsole("Test Shutdown")
#setup 3 loggers:
#log debug and info to stdout
#log warning and above to stderr
#log info and above to a file
logging.getLogger().setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logging_path = 'graceful_shutdown_test.log'
rot_file_handler = RotatingFileHandler(logging_path, maxBytes=50 * 1024 * 1024, backupCount=5)
rot_file_handler.setFormatter(formatter)
rot_file_handler.setLevel(logging.INFO)
logging.getLogger().addHandler(rot_file_handler)
log_to_stdout = logging.StreamHandler(sys.stdout)
log_to_stdout.setLevel(logging.INFO)
log_to_stdout.addFilter(lambda record: record.levelno <= logging.INFO)
log_to_stdout.setFormatter(formatter)
logging.getLogger().addHandler(log_to_stdout)
log_to_stderr = logging.StreamHandler()
log_to_stderr.setLevel(logging.WARNING)
log_to_stderr.setFormatter(formatter)
logging.getLogger().addHandler(log_to_stderr)
logging.info("start shutdown test")
#init graceful shutdown with tkinter window handle
shutdown = GracefulShutdown(console.get_window_handle())
counter = 0
counterError = 0
#test cleanup function which runs if shutdown is requested
def graceful_shutdown():
logging.info("start shutdown")
time.sleep(15)
logging.info("stop shutdown")
shutdown.add_cleanup_function(graceful_shutdown)
#main test loop
while not shutdown.shutdown_requested:
console.update()
counter += 1
if counter > 50:
logging.info("still alive")
counter = 0
counterError += 1
if counterError > 150:
logging.error("error for test")
try:
raise NameError("i'm a exception")
except:
logging.exception("exception found!")
counterError = 0
time.sleep(0.1)
shutdown.exit_gracefully()