我正在寻找从多个异步任务记录日志的相似解决方案。
如上所述,无法确定哪行日志来自哪个任务。
以下是我对此问题的解决方案。
如果有缺点或更好的方法,请让我知道。
import asyncio
import logging
log_format = '%(levelname).1s,%(thread)s,%(name)s,%(lineno)d: %(message)s'
logging.basicConfig(format=log_format, level=logging.NOTSET)
log = logging.getLogger()
class MyStreamHandler(logging.StreamHandler):
def __init__(self):
super().__init__()
def emit(self, record: logging.LogRecord) -> None:
try:
task = asyncio.current_task(asyncio.get_running_loop())
if task is not None:
record.__setattr__("thread", f"{record.thread}[{task.get_name()}]")
except RuntimeError:
pass
super().emit(record)
handler = MyStreamHandler()
handler.setFormatter(logging.Formatter(log_format))
log.handlers = []
log.addHandler(handler)
def synchronous():
log.debug("I'm synchronous")
async def asynchronous():
log.debug("I'm a debug")
log.warning("I'm a warning")
log.error("I'm an error")
async def main():
loop = asyncio.get_event_loop()
tasks = []
for i in range(5):
tasks.append(asynchronous())
await asyncio.gather(*tasks)
synchronous()
asyncio.run(main())
示例输出:
D,6052,root,30: I'm synchronous
D,6052,asyncio,623: Using proactor: IocpProactor
D,6052[Task-2],root,34: I'm a debug
W,6052[Task-2],root,35: I'm a warning
E,6052[Task-2],root,36: I'm an error
D,6052[Task-3],root,34: I'm a debug
W,6052[Task-3],root,35: I'm a warning
E,6052[Task-3],root,36: I'm an error
D,6052[Task-4],root,34: I'm a debug
W,6052[Task-4],root,35: I'm a warning
E,6052[Task-4],root,36: I'm an error
D,6052[Task-5],root,34: I'm a debug
W,6052[Task-5],root,35: I'm a warning
E,6052[Task-5],root,36: I'm an error
D,6052[Task-6],root,34: I'm a debug
W,6052[Task-6],root,35: I'm a warning
E,6052[Task-6],root,36: I'm an error