还有一些方法...
示例：日志记录模块
也许您需要的只是从 stderr 切换到 Python 的logging模块，它具有强大的发布/订阅模型。
很容易开始生成日志记录。
# producer
import logging
log = logging.getLogger("myjobs") # that's all the setup you need
class MyJob(object):
def run(self):
log.info("starting job")
n = 10
for i in range(n):
log.info("%.1f%% done" % (100.0 * i / n))
log.info("work complete")
在消费者方面还有更多的工作。不幸的是，配置记录器输出需要 7 整行代码。;)
# consumer
import myjobs, sys, logging
if user_wants_log_output:
ch = logging.StreamHandler(sys.stderr)
ch.setLevel(logging.INFO)
formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s")
ch.setFormatter(formatter)
myjobs.log.addHandler(ch)
myjobs.log.setLevel(logging.INFO)
myjobs.MyJob().run()
另一方面，日志包中有大量的东西。如果您需要将日志数据发送到一组轮换的文件、一个电子邮件地址和 Windows 事件日志，那么您将得到保障。
示例：最简单的观察者
但是您根本不需要使用任何库。支持观察者的一个极其简单的方法是调用一个什么都不做的方法。
# producer
class MyJob(object):
def on_progress(self, pct):
"""Called when progress is made. pct is the percent complete.
By default this does nothing. The user may override this method
or even just assign to it."""
pass
def run(self):
n = 10
for i in range(n):
self.on_progress(100.0 * i / n)
self.on_progress(100.0)
# consumer
import sys, myjobs
job = myjobs.MyJob()
job.on_progress = lambda pct: sys.stdout.write("%.1f%% done\n" % pct)
job.run()
有时，您可以直接说 lambda 而不是编写 lambda，job.on_progress = progressBar.update这很好。
这很简单。一个缺点是它自然不支持订阅相同事件的多个侦听器。
示例：类 C# 事件
通过一些支持代码，您可以在 Python 中获取类似 C# 的事件。这是代码：
# glue code
class event(object):
def __init__(self, func):
self.__doc__ = func.__doc__
self._key = ' ' + func.__name__
def __get__(self, obj, cls):
try:
return obj.__dict__[self._key]
except KeyError, exc:
be = obj.__dict__[self._key] = boundevent()
return be
class boundevent(object):
def __init__(self):
self._fns = []
def __iadd__(self, fn):
self._fns.append(fn)
return self
def __isub__(self, fn):
self._fns.remove(fn)
return self
def __call__(self, *args, **kwargs):
for f in self._fns[:]:
f(*args, **kwargs)
生产者使用装饰器声明事件：
# producer
class MyJob(object):
@event
def progress(pct):
"""Called when progress is made. pct is the percent complete."""
def run(self):
n = 10
for i in range(n+1):
self.progress(100.0 * i / n)
#consumer
import sys, myjobs
job = myjobs.MyJob()
job.progress += lambda pct: sys.stdout.write("%.1f%% done\n" % pct)
job.run()
这与上面的“简单观察者”代码完全一样，但您可以使用+=. （与 C# 不同，没有事件处理程序类型，new EventHandler(foo.bar)订阅事件时不必这样做，也不必在触发事件之前检查 null。与 C# 一样，事件不会抑制异常。）
如何选择
如果logging满足您的所有需求，请使用它。否则，做最适合你的最简单的事情。需要注意的关键是您不需要承担很大的外部依赖。