我们通常需要使内部对象状态和状态转换更加可见。以下三种常见情况可提高可见性:
- 一种情况是审计应用程序,我们希望保存对象状态更改的历史记录。
- 另一种情况是跟踪应用程序的安全操作,并确定谁在执行敏感操作。
- 第三种常见情况是帮助调试在使用过程中出现的问题。
Python 记录器是一种使内部对象状态和状态转换可见的灵活方法
有时,我们有多个包含不同类型信息的日志。我们可以将安全性、审计和调试分发到单独的日志中。在其他情况下,我们可能需要一个统一的日志。logging
模块允许多种配置。
一些用户可能需要详细的输出,以确认程序按照他们理解的方式工作。允许他们设置详细级别会产生各种各样的日志详细信息,重点关注用户的需求。
warnings
模块还可以为开发者和用户提供有用的信息,包括:
- 对于开发人员,我们可以使用警告向他们显示 API 已被弃用
- 对于用户,我们可能希望向他们展示结果是有问题的,但不是错误的
可能存在值得怀疑的假设,或者可能会混淆应向用户指出的默认值。
软件维护人员需要有选择地启用日志以执行有用的调试。我们很少需要覆盖调试输出:生成的日志可能非常密集。我们通常需要集中调试来跟踪特定类或模块中的特定问题。将多个日志发送到单个处理程序的想法可用于在某些位置启用详细日志记录,在其他位置启用摘要日志记录。Python 3.7.2 版本的标准库中的日志记录包没有完整的类型提示。因此,本章中的示例没有类型详细信息。在本章中,我们将介绍以下主题:
- 创建基本日志
- 配置问题
- 用于控制、调试、审核和安全性的专门日志记录
- 使用警告模块
- 高级日志记录最后几条消息和网络目标
本章的代码文件可在上找到 https://git.io/fj2U1 。
生成日志有三个步骤。两个必要的步骤如下:
- 使用
logging.getLogger()
函数获取logging.Logger
实例;例如,logger=logging.getLogger("demo")
。 - 使用该
Logger
创建消息。有许多方法,其名称为warn()
、info()
、debug()
、error()
和fatal()
,它们可以创建具有不同重要性级别的消息。例如,logger.info("hello world")
。
然而,这两个步骤不足以给我们任何输出。当我们想要查看记录的消息时,还有第三个可选步骤。进行第三步的原因是,并不总是需要查看日志。考虑一个通常保持沉默的调试日志。可选步骤是配置logging
模块的处理程序、过滤器和格式化程序。我们可以使用logging.basicConfig()
函数进行此操作;例如,logging.basicConfig(stream=sys.stderr, level=logging.INFO)
。
从技术上讲,可以跳过第一步。我们可以使用默认的记录器,它是logging
模块顶级功能的一部分。我们在第 9 章中向您展示了这一点,装饰和混合–横切方面,因为重点是装饰,而不是日志记录。建议不要使用默认的根记录器,并建议使用根记录器的子级命名记录器更易于配置。
Logger
类的实例由 name 属性标识。名称是以点分隔的字符串,形成层次结构。有一个名为""
的根记录器,空字符串。所有其他Logger
实例都是此根Logger
实例的子实例。名为foo
的复杂应用程序可能具有名为services
的内部包,其中包含名为persistence
的模块和名为SQLStore
的类。这可能导致记录者被命名为""
、"foo"
、"foo.services"
、"foo.services.persistence"
和"foo.services.persistence.SQLStore"
我们通常可以使用根目录Logger
来配置Logger
实例的整个树。当我们选择名称以形成适当的层次结构时,我们可以通过配置适当的父Logger
对象来启用或禁用相关实例的整个子树。在前面的示例中,我们可能会启用对"foo.services.persistence"
的调试,以查看来自所有类的消息,这些类的记录器名称都有一个公共前缀。
除了名称之外,每个Logger
对象还可以配置一个Handler
实例列表,用于确定消息写入的位置,以及一个Filter
对象列表,用于确定哪些类型的消息被传递或拒绝。这些Logger
实例具有日志记录的基本 API;我们使用一个Logger
对象来创建LogRecord
实例。然后将这些记录路由到Filter
和Handler
对象;传递的记录被格式化,最终存储在本地文件中,或通过网络传输。
最佳实践是为我们的每个类或模块都有一个不同的记录器。由于Logger
对象名是点分隔字符串,Logger
实例名可以与类或模块名并行;我们的应用程序的组件定义层次结构将有一个并行的记录器层次结构。我们可能有一个类似以下代码的类:
import logging
class Player:
def __init__(self, bet: str, strategy: str, stake: int) -> None:
self.logger = logging.getLogger(
self.__class__.__qualname__)
self.logger.debug(
"init bet %r, strategy %r, stake %r",
bet, strategy, stake
)
self.__class__.__qualname__
的唯一值将确保用于该类的Logger
对象的名称与该类的限定名称匹配。
作为一种通用的日志记录方法,这种方法工作得很好。这种方法唯一的缺点是,每个日志记录程序实例都是作为对象的一部分创建的,这是一种微小的冗余。将记录器创建为类的一部分,而不是类的每个实例,这会更好地使用内存。
在下一节中,我们将介绍几种创建由类的所有实例共享的类级记录器的方法。
正如我们在第 9 章、Decorators and mixin–横切方面中所述,创建类级记录器可以通过 decorator 完成。这将把记录器的创建与类的其余部分分开。一个非常简单的常见装饰想法有一个隐藏的问题。以下是装饰器示例:
def logged(cls: Type) -> Type:
cls.logger = logging.getLogger(cls. __qualname__ )
return cls
@logged
装饰器将logger
属性创建为类的特征。然后,所有实例都可以共享该类。使用此装饰器,我们可以使用如下示例所示的代码定义一个类:
@logged
class Player_1:
def __init__ ( self , bet: str , strategy: str , stake: int ) -> None :
self .logger.debug( "init bet %s, strategy %s, stake %d" , bet,
strategy, stake)
这将确保Player_1
类具有预期名称为logger
的记录器。然后我们可以在这个类的各种方法中使用self.logger
。
此设计的问题是mypy
无法检测logger
实例变量的存在。此缺口将导致mypy
报告潜在问题。有几种更好的方法可以创建记录器。
我们可以使用如下示例所示的代码创建类级调试器:
class Player_2:
logger = logging.getLogger( "Player_2" )
def __init__ ( self , bet: str , strategy: str , stake: int ) -> None :
self .logger.debug( "init bet %s, strategy %s, stake %d" , bet, strategy, stake)
这很简单,也很清楚。它有一个小的不要重复自己(干燥的问题。类名在类级记录器创建中重复。这是在 Python 中创建类的方式的结果,在类存在之前创建的对象没有简单的方法提供类名。元类的工作是完成类定义的任何终结;这可以包括为内部对象提供类名。
我们可以使用以下设计在各种相关类中构建一致的日志记录属性:
class LoggedClassMeta( type ):
def __new__ ( cls , name, bases, namespace, **kwds):
result = type . __new__ ( cls , name, bases, dict (namespace))
result.logger = logging.getLogger(result. __qualname__ )
return result
class LoggedClass( metaclass =LoggedClassMeta):
logger: logging.Logger
这个元类使用__new__()
方法创建结果对象,并向类中添加一个记录器。例如,一个名为C
的类将有一个C.logger
对象。LoggedClass
可以用作 mixin 类,以提供可见的logger
属性名称,并确保其已正确初始化。
我们将使用该类,如下例所示:
class Player_3(LoggedClass):
def __init__ ( self , bet: str , strategy: str , stake: int ) -> None :
self .logger.debug(
"init bet %s, strategy %s, stake %d" ,
bet, strategy, stake)
当我们创建一个Player_3
实例时,我们将练习logger
属性。因为这个属性是由元类为LoggedClass
设置的,所以它是为Player_3
类的每个实例可靠地设置的。
元类和超类对看起来很复杂。它为每个实例创建一个共享类级记录器。类的名称在代码中不重复。客户的唯一义务是将LoggedClass
作为混入。
默认情况下,我们不会看到这样的定义的任何输出。logging
模块的初始配置不包括产生任何输出的处理程序或级别。我们还需要更改logging
配置以查看任何输出。
logging
模块工作方式最重要的好处是,我们可以在类和模块中包含日志功能,而不必担心整体配置。默认行为将是静默的,并引入很少的开销。因此,我们可以在定义的每个类中始终包含日志功能。
为了在日志中查看输出,我们需要提供以下两个配置详细信息:
- 我们正在使用的记录器需要与至少一个产生显著输出的处理程序相关联。
- 处理程序需要一个日志级别来传递日志消息。
logging
包有多种配置方法。我们将在这里向您展示logging.basicConfig()
。我们将分别看一看logging.config.dictConfig()
。
logging.basicConfig()
方法允许几个参数创建一个logging.handlers.StreamHandler
来记录输出。在许多情况下,这就是我们所需要的:
>>> import logging
>>> import sys
>>> logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
这将配置一个将写入sys.stderr
的StreamHandler
实例。它将传递级别大于或等于给定级别的消息。通过使用logging.DEBUG
,我们可以确保看到所有消息。默认级别为logging.WARN
。
执行基本配置后,我们将在创建类时看到调试消息,如下所示:
>>> pc3 = Player_3("Bet3", "Strategy3", 3)
DEBUG:Player_3:init bet Bet3, strategy Strategy3, stake 3
默认日志格式向我们显示级别(DEBUG
)、记录器名称(Player_3
)以及我们生成的字符串。LogRecord
中有更多属性也可以添加到输出中。通常,这种默认格式是可以接受的。
logging
模块的定义方式避免了手动管理全局状态信息。全局状态在logging
模块中处理。我们可以在单独的部分编写应用程序,并且可以很好地保证这些组件将通过logging
接口进行适当的协作。例如,我们可以在一些模块中包含logging
,而在其他模块中完全省略它,而不必担心兼容性或配置。
最重要的是,我们可以在整个应用程序中包含日志记录请求,并且从不配置任何处理程序。顶级主脚本可以完全省略import logging
。在这种情况下,日志记录是一种备用功能,可以在需要调试时使用。
由于日志记录的分散性,最好只在应用程序的顶层配置一次。我们应该在应用程序的if __name__ == "__main__":
部分中配置logging
。我们将在第 18 章处理命令行中更详细地了解这一点。
我们的许多日志处理程序都涉及缓冲。在大多数情况下,数据将在正常事件过程中从缓冲区刷新。虽然我们可以忽略日志记录是如何关闭的,但使用logging.shutdown()
来确保所有缓冲区都被刷新到设备上稍微可靠一些。
在处理顶级错误和异常时,我们有两种明确的技术来确保写入所有缓冲区。一种技术是在try:
块上使用finally
子句,如下所示:
import sys
if __name__ == "__main__":
logging.config.dictConfig(yaml.load("log_config.yaml"))
try:
application = Main()
status = application.run()
except Exception as e:
logging.exception(e)
status = 1
finally:
logging.shutdown()
sys.exit(status)
本例向我们展示了如何尽早配置logging
并尽可能晚地关闭logging
。这确保了尽可能多的应用程序被正确配置的记录器正确地包围起来。这包括一个异常记录器;在某些应用程序中,main()
函数处理所有异常,使得此处的except
子句多余。
另一种方法是包括一个atexit
处理程序来关闭logging
,如下所示:
import atexit
import sys
if __name__ == "__main__":
logging.config.dictConfig(yaml.load("log_config.yaml"))
atexit.register(logging.shutdown)
try:
application = Main()
status = application.run()
except Exception as e:
logging.exception(e)
status = 2
sys.exit(status)
这个版本向我们展示了如何使用atexit
处理程序来调用logging.shutdown()
。当应用程序退出时,将调用给定的函数。如果在main()
函数中正确处理了异常,则可以用更简单的status = main(); sys.exit(status)
替换try:
块。
还有第三种技术,它使用上下文管理器来控制日志记录。我们将在第 18 章处理命令行中探讨该替代方案。
使用logging.getLogger()
命名我们的Loggers
有四种常见的用例。我们经常选择与应用程序架构并行的名称,如以下示例所述:
- 模块名称:对于包含大量创建了大量对象的小函数或类的模块,我们可能会有一个模块全局
Logger
实例。例如,当我们扩展tuple
时,我们不希望在每个实例中都引用Logger
。我们通常会在全局范围内执行此操作,通常靠近模块的前端,如下所示:
import logging
logger = logging.getLogger(__name__)
- 对象实例:这是前面我们在
__init__()
方法中创建Logger
时显示的。此Logger
将是该实例所特有的;仅使用限定的类名可能会产生误导,因为该类将有多个实例。更好的设计是在记录器的名称中包含唯一的实例标识符,如下所示:
def __init__(self, player_name)
self.name = player_name
self.logger = logging.getLogger(
f"{self.__class__.__qualname__}.{player_name}")
- 类名:这在前面我们定义一个简单的装饰器时显示过。我们可以使用
__class__.__qualname__
作为Logger
名称,并将Logger
作为一个整体分配给类。它将由类的所有实例共享。 - 函数名:对于经常使用的小函数,我们将经常使用模块级日志,如前所示。对于很少使用的较大函数,我们可以在函数中创建日志,如下所示:
def main():
log = logging.getLogger("main")
这里的想法是确保我们的Logger
名称与软件架构中的组件名称匹配。这为我们提供了最透明的日志记录,简化了调试。
然而,在某些情况下,我们可能会有一个更复杂的Loggers
集合。一个类中可能有几种不同类型的信息性消息。两个常见的例子是财务审计日志和安全访问日志。我们可能需要几个平行的Loggers
层次结构;一个名称以audit.
开头,另一个名称以security.
开头,一个类可能有更专门的Loggers
,名称如audit.module.Class
或security.module.Class
,如下例所示:
self.audit_log = logging.getLogger(
f"audit.{self.__class__.__qualname__}")
在一个类中有多个可用的记录器对象允许我们精细地控制输出的种类。我们可以将每个Logger
配置为具有不同的handlers
。我们将使用下一节中更高级的配置将输出定向到不同的目标。
logging
模块有五个预定义的重要级别。每个级别都有一个(或两个)带有级别编号的全局变量。重要性级别表示一系列可选性,从调试消息(很少重要到足以显示)到关键或致命错误(始终重要),如下表所示:
| 记录模块变量 | 值 |
| DEBUG
| 10 |
| INFO
| 20 |
| WARNING
或WARN
| 30 |
| ERROR
| 40 |
| CRITICAL
或FATAL
| 50 |
我们可以添加额外的级别,以便对传递或拒绝的消息进行更细致的控制。例如,某些应用程序支持多个级别的详细信息。类似地,一些应用程序包含多个级别的调试细节。我们可能需要添加一个额外的详细输出级别,例如设置为 15。这适用于信息和调试。它可以遵循信息性消息的模式,而无需转移到调试日志的细节。
对于普通的静默处理,我们可以将日志记录级别设置为logging.WARNING
,以便只显示警告和错误。对于第一个详细级别,我们可以设置logging.INFO
级别以查看信息性消息。对于第二个详细级别,我们可能希望添加一个值为 15 的级别,并将根记录器设置为包含此新级别。
我们可以使用以下内容定义新级别的详细消息:
logging.addLevelName(15, "VERBOSE")
logging.VERBOSE = 15
此代码需要在配置记录器之前编写。它将是顶层主脚本的一部分。我们可以通过Logger.log( )
方法使用我们的新级别,该方法将级别编号作为参数,如下所示:
self.logger.log(logging.VERBOSE, "Some Message")
虽然添加这样的级别几乎没有开销,但它们可能会被过度使用。微妙之处在于,一个级别将多个概念可见性和错误行为合并为一个数字代码。水平应限制在简单的可视性或误差范围内。任何更复杂的事情都必须通过Logger
名称或实际的Filter
对象来完成。
我们有几个将日志输出发送到多个目的地的用例,如以下项目符号列表所示:
- 我们可能需要复制日志以提高操作的可靠性。
- 我们可能正在使用复杂的
Filter
对象来创建不同的消息子集。 - 对于每个目的地,我们可能有不同的级别。我们可以使用调试级别将调试消息与信息消息分开。
- 基于
Logger
名称,我们可能有不同的处理程序来表示不同的焦点。
当然,我们也可以结合这些不同的选择来创建相当复杂的场景。为了创建多个目的地,我们必须创建多个Handler
实例。每个Handler
可能包含一个定制的Formatter
;它可以包含一个可选级别和一个可选的可应用过滤器列表。
一旦我们有多个Handler
实例,我们就可以将一个或多个Logger
对象绑定到所需的Handler
实例。一个Handler
对象可以有一个级别过滤器。使用它,我们可以有多个处理程序实例;每个都可以有一个不同的过滤器,根据级别显示不同的消息组。此外,如果我们需要比只检查严重性级别的内置过滤器更复杂的过滤,我们可以显式创建Filter
对象。
虽然我们可以通过logging
模块 API 对此进行配置,但在单独的配置文件中定义大多数日志详细信息通常更为清晰。处理此问题的一种优雅方法是对配置字典使用 YAML 表示法。然后,我们可以相对直接地使用logging.config.dictConfig(yaml.load(somefile))
加载字典。
YAML 符号比configparser
接受的符号更紧凑。Python 标准库中的logging.config
文档使用 YAML 示例,因为它们清晰明了。我们将遵循这种模式。
下面是一个配置文件示例,其中包含两个处理程序和两个记录器系列:
version: 1
handlers:
console:
class: logging.StreamHandler
stream: ext://sys.stderr
formatter: basic
audit_file:
class: logging.FileHandler
filename: data/ch16_audit.log
encoding: utf-8
formatter: basic
formatters:
basic:
style: "{"
format: "{levelname:s}:{name:s}:{message:s}"
loggers:
verbose:
handlers: [console]
level: INFO
audit:
handlers: [audit_file]
level: INFO
我们定义了两个处理程序:console
和audit_file
。console
是发送给sys.stderr
的StreamHandler
。注意,我们必须使用 URI 风格的语法ext://sys.stderr
来命名外部Python 资源。在此上下文中,“外部”是指配置文件的外部。此复杂字符串映射到sys.stderr
对象。audit_file
是将写入给定文件的FileHandler
。默认情况下,以a
模式打开文件进行追加。
我们还定义了名为basic
的格式化程序,其格式可以生成与通过basicConfig()
配置日志记录时创建的消息相匹配的消息。如果不使用此选项,dictConfig()
使用的默认格式只有消息文本。
最后,我们定义了两个顶级记录器,verbose
和audit
。verbose
实例将由顶级名称为verbose
的所有记录器使用。然后,我们可以使用Logger
名称,例如verbose.example.SomeClass
来创建一个实例,该实例是verbose
的子实例。每个记录器都有一个处理程序列表;在本例中,每个列表中只有一个元素。此外,我们还为每个记录器指定了日志记录级别。
下面是加载此配置文件的方法:
import logging.config
import yaml
config_dict = yaml.load(config)
logging.config.dictConfig(config_dict)
我们将 YAML 文本解析为dict
,然后使用dictConfig()
函数配置给定字典的日志记录。下面是一些获取日志和编写消息的示例:
verbose = logging.getLogger("verbose.example.SomeClass")
audit = logging.getLogger("audit.example.SomeClass")
verbose.info("Verbose information")
audit.info("Audit record with before and after state")
我们创建了两个Logger
对象;一个在verbose
家谱下,另一个在audit
家谱下。当我们写入verbose
记录器时,我们将在控制台上看到输出。但是,当我们向audit
记录器写入数据时,控制台上什么也看不到;该记录将转到配置中命名的文件。
当我们查看logging.handlers
模块时,我们看到了大量可以利用的处理程序。默认情况下,logging
模块使用旧式%
格式规范。这些与str.format()
方法的格式规范不同。在定义格式化程序参数时,我们使用了{
样式的格式化,这与str.format()
是一致的。
Loggers
的默认行为是日志记录从命名的Logger
向上通过所有父级Logger
实例传播到根Logger
实例。我们可能有较低级别的Loggers
具有特殊行为,并且有一个根Logger
定义所有Loggers
的默认行为。
因为日志记录会传播,根级别的记录器也会处理我们定义的较低级别Loggers
的任何日志记录。如果子日志记录器允许传播,这将导致重复的输出:首先,将有来自子日志记录器的输出,然后是日志记录传播到父日志记录器时的输出。如果我们想避免重复,当在几个级别上有处理程序时,我们必须关闭低级别记录器的传播。
我们前面的示例没有配置根级别Logger
。如果我们的应用程序的某个部分创建了一个名称不以audit.
或verbose.
开头的记录器,那么该附加记录器将不会与Handler
关联。要么我们需要更多的顶级名称,要么我们需要配置一个全面的根级别记录器。
如果我们添加一个根级别的记录器来捕获所有其他名称,那么我们必须小心传播规则。以下是对配置文件的修改:
loggers:
verbose:
handlers: [console]
level: INFO
propagate: False # Added
audit:
handlers: [audit_file]
level: INFO
propagate: False # Added
root: # Added
handlers: [console]
level: INFO
我们关闭了两个较低级别的记录器verbose
和audit
的传播。我们添加了一个新的根级别记录器。由于此记录器没有名称,因此它是作为一个单独的顶级字典完成的,名为root:
,与loggers:
条目并行。
如果我们没有在两个较低级别的记录器中关闭传播,则每个verbose
或audit
记录将被处理两次。在审计日志的情况下,实际上可能需要双重处理。审计数据将与审计文件一起进入控制台。
logging
模块的重要之处在于,我们不必对应用程序进行任何更改来优化和控制日志记录。我们几乎可以通过配置文件执行任何需要的操作。由于 YAML 是相对优雅的符号,我们可以非常简单地编码许多功能。
basicConfig()
日志记录方法小心地保存在配置之前创建的任何记录器。然而,logging.config.dictConfig()
方法的默认行为是禁用配置之前创建的任何记录器。
组装大型复杂应用程序时,我们可能会在import
过程中创建模块级记录器。主脚本导入的模块可能会在创建logging.config
之前创建记录器。此外,任何全局对象或类定义都可能在配置之前创建了记录器。
我们通常必须在配置文件中添加这样一行:
disable_existing_loggers: False
这将确保在配置之前创建的所有记录器仍将传播到由配置创建的根记录器。
测井的种类很多;我们将重点介绍以下四种类型:
- 错误和控制:应用程序的基本错误和控制会导致一个主日志,帮助用户确认程序确实在做它应该做的事情。这将包括足够的错误信息,用户可以使用这些信息更正问题并重新运行应用程序。如果用户启用详细日志记录,它将放大此主要错误,并使用其他用户友好的详细信息控制日志。
- 调试:供开发人员和维护人员使用;它可以包括相当复杂的实现细节。我们很少希望启用一揽子调试,但通常会启用特定模块或类的调试。
- 审核:这是一个正式确认,跟踪应用于数据的转换,以便我们能够确保正确完成处理。
- 安全:可以用来显示谁已经过身份验证;它可以帮助确认是否遵循了授权规则。它还可用于检测涉及重复密码失败的某些类型的攻击。
对于每种类型的日志,我们通常有不同的格式和处理要求。此外,其中一些是动态启用和禁用的。主错误和控制日志通常由非调试消息生成。我们可能有一个结构类似于以下代码的应用程序:
from collections import Counter
from Chapter_16.ch16_ex1 import LoggedClass
class Main(LoggedClass):
def __init__ ( self ) -> None :
self .counts: Counter[ str ] = collections.Counter()
def run( self ) -> None :
self .logger.info( "Start" )
# Some processing in and around the counter increments
self .counts[ "input" ] += 2000
self .counts[ "reject" ] += 500
self .counts[ "output" ] += 1500
self .logger.info( "Counts %s" , self .counts)
我们使用LoggedClass
类创建了一个记录器,其名称与类限定名称(Main
)匹配。我们已向此记录器写入信息性消息,向您显示我们的应用程序正常启动并正常完成。在本例中,我们使用Counter
来积累一些余额信息,这些信息可用于确认处理了正确数量的数据。
在某些情况下,我们会在处理结束时显示更正式的余额信息。我们可以这样做,以提供更易于阅读的显示:
for k in self .counts:
self .logger.info(
f" { k : .<16s } { self .counts[k] : >6,d } " )
此版本将在日志中的单独行上显示键和值。错误和控制日志通常使用最简单的格式;它可能只向我们显示消息文本,很少或没有额外的上下文。这样的一个formatter
可能会被用到:
formatters:
control:
style: "{"
format: "{levelname:s}:{message:s}"
此配置将formatter
与消息文本一起显示级别名称(INFO
、WARNING
、ERROR
或CRITICAL
。这就消除了许多细节,只提供了基本事实,以利于用户。我们称格式化程序为control
。
在以下代码中,我们将格式化程序与处理程序相关联:
handlers:
console:
class: logging.StreamHandler
stream: ext://sys.stderr
formatter: control
这将使用带有console
处理程序的control
格式化程序。
需要注意的是,Main
类创建时将创建记录器。这是在应用日志配置之前很久的事情。为了确保被定义为类的一部分的记录器得到正确遵守,配置必须包括以下内容:
disable_existing_loggers: False
这将保证在使用logging.config.dictConfig()
设置配置时,作为类定义一部分创建的记录器将被保留。
调试日志通常由开发人员启用,以监视正在开发的程序。它通常只关注特定的特性、模块或类。因此,我们通常会按名称启用和禁用记录器。配置文件可能会将一些记录器的级别设置为DEBUG
,而将其他记录器的级别设置为INFO,
甚至WARNING
级别。
我们经常在类中设计调试信息。事实上,我们可以使用调试功能作为类设计的一个特定质量特性。这可能意味着引入一组丰富的日志记录请求。例如,我们可能有一个复杂的计算,其中类状态是基本信息,如下所示:
from Chapter_16.ch16_ex1 import LoggedClass
class BettingStrategy(LoggedClass):
def bet( self ) -> int :
raise NotImplementedError ( "No bet method" )
def record_win( self ) -> None :
pass
def record_loss( self ) -> None :
pass
class OneThreeTwoSix(BettingStrategy):
def __init__ ( self ) -> None :
self .wins = 0
def _state( self ) -> Dict[ str , int ]:
return dict ( wins = self .wins)
def bet( self ) -> int :
bet = { 0 : 1 , 1 : 3 , 2 : 2 , 3 : 6 }[ self .wins % 4 ]
self.logger.debug(f"Bet {self._state()}; based on {bet}")
return bet
def record_win( self ) -> None :
self .wins += 1
self.logger.debug(f"Win: {self._state()}")
def record_loss( self ) -> None :
self .wins = 0
self.logger.debug(f"Loss: {self._state()}")
在这些类定义中,我们定义了一个超类BettingStrategy
,它提供了投注策略的一些特性。具体来说,此类定义了获取赌注金额、记录赢款或记录输款的方法。这个类别背后的想法是一个常见的谬误,即修改赌注可以在某种程度上减少机会游戏中的损失。
具体实现OneThreeTwoSix
创建了一个_state()
方法,该方法公开了相关的内部状态。此方法仅用于支持调试。我们避免使用self.__dict__
,因为它通常包含太多的信息,无法提供帮助。然后,我们可以在方法函数中的多个位置审核self._state
信息的更改。
在前面的许多示例中,我们依赖于记录器对%r
和%s
格式的使用。我们可能会用一条像self.logger.info("template with %r and %r", some_item, another_variable)
这样的线。这类行提供了一条消息,其中的字段在格式化程序处理之前通过过滤器。像这样的线路可以进行大量的控制
在本例中,我们使用了self.logger.debug(f"Win: {self._state()}")
,它使用了一个 f 字符串。日志包的筛选器和格式化程序不能用于对此输出提供细粒度控制。在审计和安全日志的情况下,首选由日志记录器控制的日志%
样式的格式。它允许日志过滤器以一致的方式编辑敏感信息。对于非正式日志条目,f 字符串非常方便。但是,使用 f-string 非常小心地将哪些信息放在日志中是很重要的。
调试输出通常通过编辑配置文件以在某些位置启用和禁用调试来有选择地启用。我们可能会对日志配置文件进行如下更改:
loggers:
betting.OneThreeTwoSix:
handlers: [console]
level: DEBUG
propagate: False
我们根据类的限定名称为特定类标识记录器。本例假设已经定义了一个名为console
的处理程序。此外,我们还关闭了传播,以防止调试消息复制到根记录器中。
这种设计隐含的想法是,调试不是我们希望通过简单的-D
选项或--DEBUG
选项从命令行启用的。为了执行有效的调试,我们通常希望通过配置文件启用选定的记录器。我们将在第 18 章、处理命令行中讨论命令行问题。
审计和安全日志通常在两个处理程序之间重复:主控制处理程序和用于审计和安全审查的文件处理程序。这意味着我们将执行以下操作:
- 为审计和安全定义其他记录器
- 为这些记录器定义多个处理程序
- (可选)为审核处理程序定义其他格式
如前所示,我们通常会创建audit
和security
日志的独立层次结构。创建单独的日志记录者层次结构比尝试通过新的日志记录级别引入审计或安全性要简单得多。添加新级别很有挑战性,因为这些消息本质上是INFO
消息;它们不属于INFO
的WARNING
端,因为它们不是错误,也不属于INFO
的DEBUG
端,因为它们不是可选的。
这里是对前面显示的元类的扩展。这个新元类将构建一个类,其中包括一个普通的控件或调试记录器以及一个特殊的审计记录器:
from Chapter_16.ch16_ex1 import LoggedClassMeta
class AuditedClassMeta(LoggedClassMeta):
def __new__ ( cls , name, bases, namespace, **kwds):
result = LoggedClassMeta. __new__ ( cls , name, bases, dict (namespace))
for item, type_ref in result. __annotations__ .items():
if issubclass (type_ref, logging.Logger):
prefix = "" if item == "logger" else f" { item } ."
logger = logging.getLogger(
f" { prefix }{ result. __qualname__ } " )
setattr (result, item, logger)
return result
class AuditedClass(LoggedClass, metaclass =AuditedClassMeta):
audit: logging.Logger
pass
AuditedClassMeta
定义扩展了LoggedClassMeta
。基本元类根据类名使用特定的记录器实例初始化 logged 属性。这个扩展做了类似的事情。它查找引用logging.Logger
类型的所有类型注释。所有这些引用都使用一个基于属性名和限定类名的类级记录器自动初始化。这使我们能够构建一个审计记录器或其他一些只包含类型注释的专门记录器。
AuditedClass
定义扩展了LoggedClass
定义,为类的logger
属性提供定义。这个类将audit
属性添加到类中。任何子类都将使用两个记录器创建。一个记录器的名称仅基于类的限定名称。另一个记录器使用限定名称,但前缀将其置于audit
层次结构中。下面是我们如何使用该类:
class Table(AuditedClass):
def bet( self , bet: str , amount: int ) -> None :
self .logger.info( "Betting %d on %s" , amount, bet)
self .audit.info( "Bet:%r, Amount:%r" , bet, amount)
我们创建了一个类,该类将在具有前缀为'audit.'
的名称的记录器上生成记录。其思想是从应用程序中获得两个独立的日志记录流。在主控制台日志中,我们可能希望看到一个简化的视图,如以下示例记录:
INFO:Table:Betting 1 on Black
INFO:audit.Table:Bet:'Black', Amount:1
但是,在详细的审计文件中,我们需要更多信息,如以下示例记录所示:
INFO:audit.Table:2019-03-19 07:34:58:Bet:'Black', Amount:1
INFO:audit.Table:2019-03-19 07:36:06:Bet:'Black', Amount:1
audit.Table
记录有两种不同的处理程序。每个处理程序都有不同的格式。我们可以配置日志来处理这个额外的记录器层次结构。我们将研究我们需要的两个处理程序,如下所示:
handlers:
console:
class: logging.StreamHandler
stream: ext://sys.stderr
formatter: basic
audit_file:
class: logging.FileHandler
filename: data/ch16_audit.log
encoding: utf-8
formatter: detailed
console
处理程序具有面向用户的日志条目,使用basic
格式。audit_file
处理程序使用一个名为detailed
的更复杂的格式化程序。以下是这些handlers
引用的两个formatters
:
formatters:
basic:
style: "{"
format: "{levelname:s}:{name:s}:{message:s}"
detailed:
style: "{"
format: "{levelname:s}:{name:s}:{asctime:s}:{message:s}"
datefmt: "%Y-%m-%d %H:%M:%S"
basic
格式只显示消息的三个属性。detailed
格式规则有些复杂,因为日期格式与其他消息格式是分开进行的。datetime
模块使用%
样式格式。我们对整个消息使用了{
样式的格式。以下是两个Logger
定义:
loggers:
audit:
handlers: [console,audit_file]
level: INFO
propagate: True
root:
handlers: [console]
level: INFO
我们为audit
层次结构定义了一个记录器。audit
的所有孩子都将向console Handler
和audit_file Handler
发送信息。根记录器将定义所有其他记录器仅使用控制台。现在我们将看到两种形式的审核消息。
重复处理程序在主控制台日志的上下文中为我们提供审计信息,并在一个单独的日志中提供一个可保存以供以后分析的重点审计跟踪
面向对象开发通常涉及对类或模块执行重要的重构。在我们第一次编写应用程序时,很难让 API 完全正确。实际上,使 API 完全正确所需的设计时间可能会被浪费。Python 的灵活性允许我们在进行更改时有很大的自由度,因为我们了解了更多关于问题域和用户需求的信息。
我们可以用来支持设计演进的工具之一是warnings
模块。warnings
有以下两个明确的用例和一个模糊的用例:
- 警告应该用来提醒开发者 API 的变化;通常,已弃用或待弃用的功能。默认情况下,弃用和挂起的弃用警告是无提示的。运行
unittest
模块时,这些消息不是无声的;这有助于我们确保正确使用升级的库包。 - 警告应提醒用户配置问题。例如,一个模块可能有几个替代实现;当首选实现不可用时,我们可能希望提供一个警告,说明没有使用最佳实现。
- 我们可能会使用警告来提醒用户,计算结果可能有问题,从而打破界限。在 Python 环境之外,警告的一个定义是,……表示服务可能已经执行了部分但不是全部请求的函数。关于不完整的结果导致警告的想法存在争议:与其产生警告,不如不产生结果潜在的不完整结果。
对于前两个用例,我们将经常使用 Python 的warnings
模块向您展示存在可纠正的问题。对于第三个模糊的用例,我们可以使用logger.warn()
方法提醒用户潜在的问题。我们不应该依赖于warnings
模块,因为默认行为是只显示一次警告。
警告模块的价值在于提供可选的消息,旨在优化、兼容性和一小部分运行时问题。例如,使用复杂库或包的实验功能可能会导致警告
当我们更改某个模块、包或类的 API 时,我们可以通过warnings
模块提供一个方便的标记。这将在已弃用或挂起弃用的方法中引发警告,如下所示:
import warnings
class Player:
"""version 2.1"""
def bet( self ) -> None :
warnings.warn(
"bet is deprecated, use place_bet" ,
DeprecationWarning , stacklevel = 2 )
pass
当我们这样做时,应用程序中使用Player.bet()
的任何部分都将收到DeprecationWarning
。默认情况下,此警告是无声的。但是,我们可以调整warnings
过滤器以查看消息,如下所示:
>>> warnings.simplefilter("always", category=DeprecationWarning)
>>> p2 = Player()
>>> p2.bet() __main__:4: DeprecationWarning: bet is deprecated, use
place_bet
这种技术允许我们定位由于 API 更改而必须更改应用程序的所有位置。如果我们有接近 100%代码覆盖率的单元测试用例,那么这个简单的技术很可能揭示所有不推荐方法的用法。
一些集成开发环境(IDE)可以发现警告的使用并突出显示不推荐使用的代码。例如,PyCharm 将通过使用任何不推荐的bet()
方法来划定一条小界线。
因为这对于规划和管理软件更改非常有价值,我们有以下三种方法使警告在应用程序中可见:
- 命令行
-Wd
选项将所有警告的操作设置为default
。这将启用通常无提示的弃用警告。当我们运行python3.7 -Wd
时,我们将看到所有的弃用警告。 - 使用
unittest
,始终在warnings.simplefilter('default')
模式下执行。 - 包括
warnings.simplefilter('default')
在我们的应用程序中。这也将对所有警告应用default
操作;它相当于-Wd
命令行选项。
对于给定的类或模块,我们可能有多个实现。我们通常会使用配置文件参数来决定哪个实现是合适的。有关此技术的更多信息,请参见第 14 章、配置文件和持久化。
但是,在某些情况下,应用程序可能会静默地依赖于其他包是否是 Python 安装的一部分。一种实现可能是最优的,另一种实现可能是后备计划。许多 Python 库模块使用它在优化的二进制模块和纯 Python 模块之间进行选择。
一种常见的技术是尝试多个import
选项来定位已安装的软件包。我们可以生成警告,显示可能的配置困难。以下是管理此替代实施导入的方法:
import warnings
try :
import simulation_model_1 as model
except ImportError as e:
warnings.warn( repr (e))
if 'model' not in globals ():
try :
import simulation_model_2 as model
except ImportError as e:
warnings.warn( repr (e))
if 'model' not in globals():
raise ImportError("Missing simulation_model_1 and simulation_model_2")
我们尝试了一个模块的导入。如果失败了,我们会尝试另一次导入。我们使用了一个if
语句来减少异常的嵌套。如果有两个以上的备选方案,嵌套异常可能导致外观非常复杂的异常。通过使用额外的if
语句,我们可以将一长串备选方案展平,这样就不会嵌套异常。
通过更改消息的类别,我们可以更好地管理此警告消息。在前面的代码中,这将是UserWarning
。默认情况下会显示这些选项,为用户提供一些证据,证明配置不是最优的。
如果我们将类更改为ImportWarning
,则默认为静默。在包的选择与用户无关的情况下,这提供了一个正常的静默操作。典型的开发人员使用-Wd
选项运行的技术将显示ImportWarning
消息。
要更改警告的级别,我们将调用更改为warnings.warn()
,如下所示:
warnings.warn(e, ImportWarning)
这会将警告更改为默认为静默的类。开发人员仍然可以看到该消息,他们应该使用-Wd
选项。
针对终端用户发出警告的想法有点模糊;申请成功了还是失败了?警告的真正含义是什么?用户是否应该做一些不同的事情?
由于这种潜在的模糊性,用户界面中的警告不是一个好主意。要真正可用,程序要么正常工作,要么根本不工作。当出现错误时,错误消息应该包括关于用户对问题的响应的建议。我们不应该让用户承担判断输出质量和确定其适用性的负担。我们将详细阐述这一点。
A program should either work correctly or it should not work at all*.*
最终用户警告的一个潜在明确用途是提醒用户输出不完整。例如,应用程序可能在完成网络连接时遇到问题;基本结果是正确的,但其中一个数据源工作不正常。
在某些情况下,应用程序执行的操作不是用户所请求的,并且输出是有效和有用的。在出现网络问题的情况下,尽管存在网络问题,仍可以使用默认行为。一般来说,用正确但不完全符合用户要求的东西替换有故障的东西是一个很好的警告候选者。此类警告最好在警告级别使用logging
,而不是使用warnings
模块。warnings
模块产生一次性消息;我们可能希望向用户提供更多详细信息。下面是我们如何使用一条简单的Logger.warn()
消息来描述日志中的问题:
try:
with urllib.request.urlopen("http://host/resource/", timeout=30) as resource:
content = json.load(resource)
except socket.timeout as e:
self.log.warn(
"Missing information from http://host/resource")
content= []
如果发生超时,将向日志中写入警告消息,程序将继续运行。资源的内容将设置为空列表。每次都会写入日志消息。warnings
模块警告通常仅在程序中的给定位置显示一次,然后被抑制。
我们将研究两种更高级的技术,它们可以帮助提供有用的调试信息。第一个是一条*木尾巴;*这是某个重要事件之前最后几条日志消息的缓冲区。这个想法是要有一个可以读取的小文件来显示应用程序为什么会死掉。这有点像 OStail
命令自动应用于完整日志输出。
第二种技术使用日志框架的一项功能,通过网络将日志消息发送到集中式日志处理服务。这可用于整合来自多个并行 web 服务器的日志。我们需要为日志创建发送方和接收方。
日志尾缓冲区是对logging
框架的扩展。我们将扩展MemoryHandler
来稍微改变它的行为。MemoryHandler
的内置行为包括三个写入用例,当容量达到时,它将写入另一个handler
;当logging
关闭时,将写入任何缓冲消息;最重要的是,当记录给定级别的消息时,它将写入整个缓冲区。
我们将稍微更改第一个用例。我们将删除最旧的消息,而不是在缓冲区已满时写入输出文件,将其他消息留在缓冲区中。在退出时写入和在处理高严重性记录时写入的另外两个用例将被单独处理。这将导致在关机前转储最后几条消息,以及在出错前转储最后几条消息。
MemoryHandler
实例的默认配置是缓冲消息,直到记录大于或等于ERROR
级别的消息。这将导致在记录错误时转储缓冲区。它将倾向于使不是错误直接前兆的业务照常消息保持沉默。
为了理解这个示例,定位 Python 安装并详细查看logging.handlers
模块非常重要。
MemoryHandler
的此扩展将根据TailHandler
类创建时定义的容量保留最后几条消息,如下所示:
class TailHandler(logging.handlers.MemoryHandler):
def shouldFlush( self , record: logging.LogRecord) -> bool :
"""
Check for buffer full
or a record at the flushLevel or higher.
"""
if record.levelno >= self .flushLevel:
return True
while len ( self .buffer) > self .capacity:
self .acquire()
try :
del self .buffer[ 0 ]
finally :
self .release()
return False
我们扩展了MemoryHandler
,以便它将日志消息累积到给定的容量。当达到容量时,旧邮件将在添加新邮件时被删除。注意,我们必须锁定数据结构以允许多线程日志记录。
如果接收到具有适当级别的消息,则整个结构将发送到目标处理程序。通常,目标是FileHandler
,它会写入尾部文件以进行调试和支持。
此外,当日志记录关闭时,最后几条消息也将写入尾部文件。这应该表示不需要任何调试或支持的正常终止。
通常,我们会将DEBUG
级别的消息发送给此类处理程序,以便我们了解大量有关崩溃情况的详细信息。配置应特别将级别设置为DEBUG
,而不是允许默认级别。
以下是使用此TailHandler
的配置:
version: 1
disable_existing_loggers: False
handlers:
console:
class: logging.StreamHandler
stream: ext://sys.stderr
formatter: basic
tail:
(): __main__.TailHandler
target: cfg://handlers.console
capacity: 5
formatters:
basic:
style: "{"
format: "{levelname:s}:{name:s}:{message:s}"
loggers:
test:
handlers: [tail]
level: DEBUG
propagate: False
root:
handlers: [console]
level: INFO
TailHandler
的定义向我们展示了logging
配置的几个附加特性。它向我们显示了类引用以及配置文件的其他元素。
我们在配置中引用了自定义类定义。标签()
指定该值应解释为模块和类名。在本例中,它是我们__main__.TailHandler
类的一个实例。标签class
而不是()
使用了属于logging
包的模块和类。
我们引用了配置中定义的另一个记录器。cfg://handlers.console
是指本配置文件handlers
部分中定义的console
处理程序。出于演示目的,我们使用了StreamHandler
尾部目标,它使用sys.stderr
。如前所述,更好的设计可能是使用以调试文件为目标的FileHandler
。
我们创建了使用tail
处理程序的test
记录器层次结构。写入这些记录器的消息将被缓冲,并且仅在错误或关机时显示。
下面是一个演示脚本:
logging.config.dictConfig(yaml.load(config8))
log = logging.getLogger( "test.demo8" )
log.info( "Last 5 before error" )
for i in range ( 20 ):
log.debug( f"Message { i : d } " )
log.error( "Error causes dump of last 5" )
log.info( "Last 5 before shutdown" )
for i in range ( 20 , 40 ):
log.debug( f"Message { i : d } " )
log.info( "Shutdown causes dump of last 5" )
logging.shutdown()
在发生错误之前,我们生成了 20 条消息。然后,在关闭日志记录和刷新缓冲区之前,我们又生成了 20 条消息。这将产生如下输出:
DEBUG:test.demo8:Message 15
DEBUG:test.demo8:Message 16
DEBUG:test.demo8:Message 17
DEBUG:test.demo8:Message 18
DEBUG:test.demo8:Message 19
ERROR:test.demo8:Error causes dump of last 5
DEBUG:test.demo8:Message 36
DEBUG:test.demo8:Message 37
DEBUG:test.demo8:Message 38
DEBUG:test.demo8:Message 39
INFO:test.demo8:Shutdown causes dump of last 5
中间消息被TailHandler
对象无声地丢弃。当容量设置为 5 时,将显示错误(或关机)之前的最后 5 条消息。最后五条消息包括四条调试消息和最后一条信息消息。
一种高性能设计模式是拥有一组用于解决单个问题的进程。我们可能有一个跨多个应用程序服务器或多个数据库客户端的应用程序。对于这种体系结构,我们通常希望在所有不同的进程之间有一个集中的日志。
创建统一日志的一种技术是包含准确的时间戳,然后将不同日志文件中的记录排序为单个统一日志。这种排序和合并是可以避免的额外处理。另一种响应性更强的技术是从多个并发生产者进程向单个消费者进程发送日志消息。
我们的共享日志解决方案利用了来自multiprocessing
模块的共享队列。有关多处理的更多信息,请参见第 13 章、发送和共享对象。
以下是构建多处理应用程序的三步过程:
- 首先,我们将创建一个生产者和消费者共享的队列对象。
- 其次,我们将创建使用者进程,该进程从队列中获取日志记录。日志记录使用者可以对消息应用过滤器并将其写入统一文件。
- 第三,我们将创建生产者进程池,这些进程执行应用程序的实际工作,并在与使用者共享的队列中生成日志记录。
作为附加功能,ERROR
和FATAL
消息可以通过短信或电子邮件向相关用户提供即时通知。使用者还可以处理与旋转日志文件相关的(相对)缓慢处理。
以下是消费者流程的定义:
import collections
import logging
import multiprocessing
class Log_Consumer_1(multiprocessing.Process):
def __init__ ( self , queue):
self .source = queue
super (). __init__ ()
logging.config.dictConfig(yaml.load(consumer_config))
self .combined = logging.getLogger(f"combined.{self.__class__.__qualname__}")
self .log = logging.getLogger( self .__class__. __qualname__ )
self .counts = collections.Counter()
def run( self ):
self .log.info( "Consumer Started" )
while True :
log_record = self .source.get()
if log_record == None :
break
self .combined.handle(log_record)
self.counts[log_record.getMessage()] += 1
self .log.info( "Consumer Finished" )
self .log.info( self .counts)
这个过程是multiprocessing.Process
的一个子类。Process 类提供了一个start()
方法,它将派生一个子进程并执行这里提供的run()
方法。
self.counts 对象跟踪来自生产者的单个消息。这里的想法是创建一个摘要,显示收到的消息的类型。这不是常见的做法,但它有助于揭示演示的工作原理。
进程运行时,此对象将使用Queue.get()
方法从队列中获取日志记录。消息将路由到记录器实例。在本例中,我们将创建一个名为combined.
的特殊记录器;这将从源进程中的每条记录中获得。
哨兵对象None
将用于发出处理结束的信号。收到此消息后,while
语句将完成,并写入最终日志消息。self.counts
对象将显示看到了多少条消息。这使我们可以调整队列大小,以确保消息不会由于队列溢出而丢失。
以下是此流程的logging
配置文件:
version: 1
disable_existing_loggers: False
handlers:
console:
class: logging.StreamHandler
stream: ext://sys.stderr
formatter: basic
formatters:
basic:
style: "{"
format: "{levelname:s}:{name:s}:{message:s}"
loggers:
combined:
handlers: [console]
formatter: detail
level: INFO
propagate: False
root:
handlers: [console]
level: INFO
我们用基本格式定义了一个简单的控制台Logger
。我们还定义了日志记录器层次结构的顶层,其名称以combined.
开头。这些记录器将用于显示各生产商的综合产量。
下面是一位日志制作人:
import multiprocessing
import time
import logging
import logging.handlers
class Log_Producer(multiprocessing.Process):
handler_class = logging.handlers.QueueHandler
def __init__ ( self , proc_id, queue):
self .proc_id = proc_id
self .destination = queue
super (). __init__ ()
self .log = logging.getLogger(
f" { self .__class__. __qualname__ } . { self .proc_id } " )
self .log.handlers = [ self .handler_class( self .destination)]
self .log.setLevel(logging.INFO)
def run( self ):
self .log.info( f"Started" )
for i in range ( 100 ):
self .log.info( f"Message { i : d } " )
time.sleep(0.001)
self .log.info( f"Finished" )
生产商在配置方面做得不多。它让记录器使用限定的类名和实例标识符(self.proc_id
。它将处理程序列表设置为仅围绕目标(一个Queue
实例)包装。此记录器的级别设置为INFO
。
我们将handler_class
作为类定义的一个属性,因为我们计划更改它。对于第一个示例,它将是logging.handlers.QueueHandler
。它允许示例生成器与其他类型的处理程序一起重用。
实际执行此工作的过程使用记录器创建日志消息。这些消息将排队等待集中式使用者处理。在这种情况下,该过程只是以尽可能快的速度向队列中发送 102 条消息。
以下是我们如何启动消费者和生产者。我们将以小组步骤展示这一点。首先,我们按如下方式创建队列:
import multiprocessing
queue= multiprocessing.Queue(10)
此队列太小,无法在几分之一秒内处理 10 条消息。小队列的概念是查看消息丢失时会发生什么。以下是我们如何启动消费者流程:
consumer = Log_Consumer_1(queue)
consumer.start()
以下是我们如何启动一系列生产者流程:
producers = []
for i in range(10):
proc= Log_Producer(i, queue)
proc.start()
producers.append(proc)
正如预期的那样,10 个并发生产者将使队列溢出。每个生产者都会收到大量异常队列,向我们显示消息丢失。
以下是我们如何干净利落地完成处理:
for p in producers:
p.join()
queue.put(None)
consumer.join()
首先,我们等待每个生产者进程完成,然后重新加入父进程。然后,我们将一个 sentinel 对象放入队列中,以便使用者可以干净地终止。最后,我们等待使用者进程完成并加入父进程。
日志模块的默认行为是使用Queue.put_nowait()
方法将消息放入队列。这样做的好处是,它允许生产者在没有与日志记录相关的延迟的情况下运行。这样做的缺点是,如果队列太小,无法处理非常大的日志消息突发,则消息将丢失。
我们有以下两种选择来优雅地处理突发消息:
- 我们可以从
Queue
切换到SimpleQueue
。SimpleQueue
的大小不确定。由于它的 API 略有不同,我们需要将QueueHandler
扩展为使用Queue.put()
而不是Queue.put_nowait()
。 - 在极少数情况下,队列已满,我们可以减慢生产商的速度。这是对
QueueHandler
的一个小改动,用Queue.put()
代替Queue.put_nowait()
。
有趣的是,相同的 API 更改适用于Queue
和SimpleQueue
。这是零钱:
class WaitQueueHandler(logging.handlers.QueueHandler):
def enqueue( self , record):
self .queue.put(record)
我们将enqueue()
方法的主体替换为Queue
的不同方法。现在,我们可以使用SimpleQueue
或Queue
。如果我们使用Queue
,它将等待队列已满,以防止日志消息丢失。如果我们使用SimpleQueue
,队列将自动扩展以容纳所有消息。
以下是修订后的制片人类别:
class Log_Producer_2(Log_Producer):
handler_class = WaitQueueHandler
本课程使用我们的新WaitQueueHandler
。否则,生产者与上一版本相同。
创建Queue
和启动使用者的脚本的其余部分是相同的。生产者是Log_Producer_2
的实例,但除此之外,要启动和加入的脚本与第一个示例相同。
此变体运行速度较慢,但不会丢失任何消息。我们可以通过创建更大的队列容量来提高性能。如果我们创建一个容量为 1020 条消息的队列,那么性能将最大化,因为这是突发的最大可能大小。找到最佳队列容量需要一些实验。确切的大小取决于操作系统,例如,30 的大小不会丢失太多消息。生产者和消费者的相对表现很重要。要查看效果,请将生成器中的睡眠时间更改为较大或较小的数字。此外,将生产商的数量从 10 个改为 100 个,这也有助于实验。
我们了解了如何使用日志模块和更先进的面向对象设计技术。我们创建了与模块、类、实例和函数相关联的日志。我们使用 decorator 创建日志记录,作为跨多个类定义的一致横切方面。
我们看到了如何使用warnings
模块向您显示配置或不推荐的方法存在问题。我们可以将警告用于其他目的,但我们需要小心过度使用警告,并在不清楚应用程序是否正常工作的情况下创建模糊的情况。
logging
模块支持可审核性和调试能力,并支持一些安全要求。我们可以使用日志作为保存处理步骤记录的简单方法。通过有选择地启用和禁用日志记录,我们可以支持那些试图了解代码在处理真实数据时真正在做什么的开发人员。
warnings
模块支持调试能力和可维护性。我们可以使用警告提醒开发人员 API 问题、配置问题和其他潜在的 bug 来源。
在使用logging
模块时,我们通常会创建大量不同的记录器,为少数handlers
提供数据。我们可以使用Logger
名称的层次性来引入新的或专门的日志消息集合。一个类没有理由不能有两个记录器:一个用于审计,另一个用于更通用的调试。
我们可以引入新的日志级别号,但这应该是不情愿的。级别倾向于将开发人员重点(调试、信息和警告)与用户重点(信息、错误和致命)混为一谈。有一种来自调试消息的可选性频谱,致命错误消息不需要它,它永远不应该被沉默。我们可能会为详细信息或可能的详细调试添加一个级别,但这就是所有应该使用级别来完成的事情。
logging
模块允许我们为不同目的提供多个配置文件。作为开发人员,我们可以使用一个配置文件,将日志记录级别设置为DEBUG
,并为正在开发的模块启用特定的日志记录程序。对于最终部署,我们可以提供一个配置文件,将日志记录级别设置为INFO
,并提供不同的处理程序来支持更正式的审计或安全审查需求。
我们将包括来自Python 的**禅宗(的一些想法 https://www.python.org/dev/peps/pep-0020/ :
"Errors should never pass silently. Unless explicitly silenced."
warnings
和logging
模块直接支持这一想法。
这些模块更多地面向总体质量,而不是问题的具体解决方案。它们允许我们通过相当简单的编程提供一致性。随着我们面向对象的设计变得越来越大、越来越复杂,我们可以更多地关注正在解决的问题,而不必在基础设施方面浪费时间。此外,这些模块允许我们定制输出,以提供开发人员或用户所需的信息。
在接下来的章节中,我们将介绍可测试性设计,以及如何使用unittest
、doctest
和pytest
包进行测试。自动化测试至关重要;除非有自动单元测试提供充分的证据证明代码是有效的,否则任何程序都不应该被认为是完整的。我们还将研究使软件更易于测试的面向对象设计技术。