python - Python 中的线程安全警告

Question

我正在尝试找到一种记录警告消息的好方法，但要在其中附加只有函数调用者知道的信息。

我想举个例子就很清楚了。

# log method as parameter

class Runner1(object):

    def __init__(self, log):
        self.log = log

    def run(self):
        self.log('First Warning')
        self.log('Second Warning')
        return 42


class Main1(object):

    def __init__(self):
        self._runner = Runner1(self.log)

    def log(self, message):
        print('Some object specific info: {}'.format(message))

    def run(self):
        print(self._runner.run())

e1 = Main1()
e1.run()

Main 对象有一个日志功能，可以在记录之前向任何消息添加自己的信息。此日志函数作为参数提供(在本例中为 Runner 对象)。一直携带这个额外的参数非常烦人，我想避免它。通常有很多对象/函数，因此我放弃了使用日志记录方法，因为我需要为每个对象创建不同的记录器。 (这是正确的吗？)

我尝试使用警告模块冒泡警告:

# warning module

import warnings

class Runner2(object):

    def run(self):
        warnings.warn('First Warning')
        warnings.warn('Second Warning')
        return 42


class Main2(object):

    def __init__(self):
        self._runner = Runner2()

    def log(self, message):
        print('Some object specific info: {}'.format(message))

    def run(self):
        with warnings.catch_warnings(record=True) as ws:
            warnings.simplefilter("always")
            out = self._runner.run()
            for w in ws:
                self.log(w.message)
        print(out)

e2 = Main2()
e2.run()

但是根据文档，这不是线程安全的。

最后，我也尝试了一些生成器:

# yield warning

class _Warning(object):

    def __init__(self, message):
        self.message = message


class Runner3(object):

    def run(self):
        yield _Warning('First Warning')
        yield _Warning('Second Warning')
        yield 42


class Main3(object):

    def __init__(self):
        self._runner = Runner3()

    def log(self, message):
        print('Some object specific info: {}'.format(message))

    def run(self):
        for out in self._runner.run():
            if not isinstance(out, _Warning):
                break
            self.log(out.message)
        print(out)


e3 = Main3()
e3.run()

但是您必须修改 Runner.run 以产生(而不是返回)最终结果这一事实很不方便，因为必须专门更改函数才能以这种方式使用(也许将来会改变？ PEP255 中的最后一次质量检查)。另外，我不确定这种实现方式是否还有其他问题。

所以我要寻找的是一种不需要传递参数的线程安全的冒泡警告方式。我也希望没有警告的方法保持不变。添加一个特殊的构造，例如 yield 或 warning.warn 来冒泡警告就可以了。

有什么想法吗？

Answer 1

import Queue

log = Queue.Queue()
class Runner1(object):

    def run(self):
        log.put('First Warning')
        log.put('Second Warning')
        return 42

class Main1(object):

    def __init__(self):
        self._runner = Runner1()

    def log(self, message):
        print('Some object specific info: {0}'.format(message))

    def run(self):
        out=self._runner.run()
        while True:
            try:
                msg = log.get_nowait()
                self.log(msg)
            except Queue.Empty:
                break
        print(out)


e1 = Main1()
e1.run()

产量

Some object specific info: First Warning
Some object specific info: Second Warning
42