15.11. atexit — 程序关闭时回调

本节目标:学习如何在程序退出时调用某些函数。
atexit 模块提供一个用于在程序正常退出时注册回调函数的接口。

注册回调函数

下面是一个调用 register() 来注册回调函数的例子。

atexit_simple.py

import atexit

def all_done():
    print('all_done()')

print('Registering')
atexit.register(all_done)
print('Registered')

我们也没写其他的事情,all_done() 马上就被调用了。

$ python3 atexit_simple.py

Registering
Registered
all_done()

我们可以注册多个函数,同时也可以传递参数。使用此种方法可以有效清理断开的数据库,清理临时文件等等。我们无需维护一个需要释放资源的列表,给每个资源分配一个清理函数即可。

atexit_multiple.py

import atexit

def my_cleanup(name):
    print('my_cleanup({})'.format(name))

atexit.register(my_cleanup, 'first')
atexit.register(my_cleanup, 'second')
atexit.register(my_cleanup, 'third')

注册的函数会以反序调用。同时也会以反序让模块从被导入的地方被清理,需要注意减少依赖冲突。

$ python3 atexit_multiple.py

my_cleanup(third)
my_cleanup(second)
my_cleanup(first)

装饰器语法

不需要参数的函数可以使用 register() 装饰器注册。使用这种方式让清理函数更方便清理模块级别的全局数据。

atexit_decorator.py

import atexit

@atexit.register
def all_done():
    print('all_done()')

print('starting main program')

这种方式是在函数定义之初就注册了,我们要确保即使没有其他工作也会让此函数正确运行。因为如果资源并未被初始化的化,此函数调用时也不会发生任何错误。

$ python3 atexit_decorator.py

starting main program
all_done()

取消回调

要取消一个退出回调,我们需要在注册表中把它删除,使用 unregister() 来完成。

atexit_unregister.py

import atexit

def my_cleanup(name):
    print('my_cleanup({})'.format(name))

atexit.register(my_cleanup, 'first')
atexit.register(my_cleanup, 'second')
atexit.register(my_cleanup, 'third')

atexit.unregister(my_cleanup)

所有相同的回调都会被取消,无论注册过多少次。

$ python3 atexit_unregister.py

尝试移除一个并未注册过的回调并不会引发异常。

atexit_unregister_not_registered.py

import atexit

def my_cleanup(name):
    print('my_cleanup({})'.format(name))

if False:
    atexit.register(my_cleanup, 'never registered')

atexit.unregister(my_cleanup)

也正由于它会替我们忽略了未知的回调的特性,unregister()可以用在未知的注册序列中尝试注销回调。.

$ python3 atexit_unregister_not_registered.py

atexit 回调有不被调用的时候吗?

atexit 所注册的回调会在以下几种情况下不被调用:

  • 程序在某信号下死掉。
  • 直接调用了 os._exit().
  • 解释器中发现了一个致命错误。

我们调整下 subprocess  章节的一个例子来展示下第一种情况。我们来调用两个文件,一个父程序,一个子程序。父程序会开启子程序,暂停一会最后杀死子程序。

atexit_signal_parent.py

import os
import signal
import subprocess
import time

proc = subprocess.Popen('./atexit_signal_child.py')
print('PARENT: Pausing before sending signal...')
time.sleep(1)
print('PARENT: Signaling child')
os.kill(proc.pid, signal.SIGTERM)

子程序中写好 atexit 回调,然后进入睡眠等待信号发生。

atexit_signal_child.py

import atexit
import time
import sys

def not_called():
    print('CHILD: atexit handler should not have been called')

print('CHILD: Registering atexit handler')
sys.stdout.flush()
atexit.register(not_called)

print('CHILD: Pausing to wait for signal')
sys.stdout.flush()
time.sleep(5)

下面是运行后的输出

$ python3 atexit_signal_parent.py

CHILD: Registering atexit handler
CHILD: Pausing to wait for signal
PARENT: Pausing before sending signal...
PARENT: Signaling child

子程序并未打印 not_called() 中的消息。

如果程序调用了 os._exit(),也会跳过 atexit 的回调。

atexit_os_exit.py

import atexit
import os

def not_called():
    print('This should not be called')

print('Registering')
atexit.register(not_called)
print('Registered')

print('Exiting...')
os._exit(0)

这样做会跳过正常的退出途径,所以回调就不能工作了。打印输出也不会刷新,所以我们要用 -u 模式启用非缓冲 I/O。

$ python3 -u atexit_os_exit.py

Registering
Registered
Exiting...

要确保回调函数正常运行,一是让程序正常运行完,或者调用 sys.exit() 也可以。

atexit_sys_exit.py

import atexit
import sys

def all_done():
    print('all_done()')

print('Registering')
atexit.register(all_done)
print('Registered')

print('Exiting...')
sys.exit()

调用 sys.exit() 就会调用回调了。

$ python3 atexit_sys_exit.py

Registering
Registered
Exiting...
all_done()

处理异常

atexit 回调所产生的错误会打印在控制台,最近一次发生的错误会被重新抛出。

atexit_exception.py

import atexit

def exit_with_exception(message):
    raise RuntimeError(message)

atexit.register(exit_with_exception, 'Registered first')
atexit.register(exit_with_exception, 'Registered second')

执行顺序取决于注册顺序。注意不要让某回调中发生的错误引起另一个回调中的错误(越早注册的回调越晚调用),如果发生了那最终的错误信息就没有什么用。

$ python3 atexit_exception.py

Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "atexit_exception.py", line 11, in exit_with_exception
    raise RuntimeError(message)
RuntimeError: Registered second
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "atexit_exception.py", line 11, in exit_with_exception
    raise RuntimeError(message)
RuntimeError: Registered first

在应用程序退出时存在大量错误会很麻烦,最好的的做法是清理函数能处理这些异常并把所有的异常写入日志。

参阅

本文章首发在 LearnKu.com 网站上。

本译文仅用于学习和交流目的,转载请务必注明文章译者、出处、和本文链接
我们的翻译工作遵照 CC 协议,如果我们的工作有侵犯到您的权益,请及时联系我们。
上一篇 下一篇
讨论数量: 0
发起讨论 只看当前版本


暂无话题~