现在我在框架中有一个中心模块，它使用Python 2.6 multiprocessing模块生成多个进程。因为它使用多处理，所以有一个模块级的多处理感知日志，log = multiprocessing.get_logger()。根据文档，这个日志记录器(EDIT)没有进程共享锁，所以你不会在sys. exe中弄乱东西。Stderr(或任何文件句柄)，让多个进程同时写入它。
我现在遇到的问题是框架中的其他模块不支持多处理。在我看来，我需要让这个中心模块上的所有依赖都使用多处理感知日志。这在框架内很烦人，更不用说对框架的所有客户端了。还有我想不到的选择吗?
当前回答
如何将所有日志记录委托给另一个进程，从队列中读取所有日志条目?
LOG_QUEUE = multiprocessing.JoinableQueue()
class CentralLogger(multiprocessing.Process):
def __init__(self, queue):
multiprocessing.Process.__init__(self)
self.queue = queue
self.log = logger.getLogger('some_config')
self.log.info("Started Central Logging process")
def run(self):
while True:
log_level, message = self.queue.get()
if log_level is None:
self.log.info("Shutting down Central Logging process")
break
else:
self.log.log(log_level, message)
central_logger_process = CentralLogger(LOG_QUEUE)
central_logger_process.start()
只需通过任何多进程机制甚至继承共享LOG_QUEUE，就可以很好地工作!
其他回答
通过使用处理程序，当前所有解决方案都与日志记录配置过于耦合。我的解决方案具有以下架构和功能:
您可以使用任何想要的日志记录配置 日志记录在守护进程线程中完成 使用上下文管理器安全关闭守护进程 与日志线程的通信是由多进程完成的。队列 在子进程中，日志记录。Logger(和已经定义的实例)被打补丁以将所有记录发送到队列 新:在发送到队列之前格式化回溯和消息，以防止pickle错误
带有使用示例和输出的代码可以在以下Gist中找到:https://gist.github.com/schlamar/7003737
解决这个问题的唯一方法是非侵入性的:
Spawn each worker process such that its log goes to a different file descriptor (to disk or to pipe.) Ideally, all log entries should be timestamped. Your controller process can then do one of the following: If using disk files: Coalesce the log files at the end of the run, sorted by timestamp If using pipes (recommended): Coalesce log entries on-the-fly from all pipes, into a central log file. (E.g., Periodically select from the pipes' file descriptors, perform merge-sort on the available log entries, and flush to centralized log. Repeat.)
只需将日志记录器的实例发布到某个地方。这样，其他模块和客户端就可以使用您的API来获取记录器，而不必导入multiprocessing。
我刚刚写了一个我自己的日志处理程序，它只是通过管道将所有内容提供给父进程。我只测试了十分钟，但它似乎工作得很好。
(注意:这是硬编码到RotatingFileHandler，这是我自己的用例。)
更新:@javier现在将这种方法作为Pypi上可用的包来维护-参见Pypi上的multiprocessing-logging, github上的https://github.com/jruere/multiprocessing-logging
更新:实现!
现在它使用队列来正确地处理并发，并正确地从错误中恢复。现在，我已经在生产中使用了几个月了，下面的当前版本工作起来没有问题。
from logging.handlers import RotatingFileHandler
import multiprocessing, threading, logging, sys, traceback
class MultiProcessingLog(logging.Handler):
def __init__(self, name, mode, maxsize, rotate):
logging.Handler.__init__(self)
self._handler = RotatingFileHandler(name, mode, maxsize, rotate)
self.queue = multiprocessing.Queue(-1)
t = threading.Thread(target=self.receive)
t.daemon = True
t.start()
def setFormatter(self, fmt):
logging.Handler.setFormatter(self, fmt)
self._handler.setFormatter(fmt)
def receive(self):
while True:
try:
record = self.queue.get()
self._handler.emit(record)
except (KeyboardInterrupt, SystemExit):
raise
except EOFError:
break
except:
traceback.print_exc(file=sys.stderr)
def send(self, s):
self.queue.put_nowait(s)
def _format_record(self, record):
# ensure that exc_info and args
# have been stringified. Removes any chance of
# unpickleable things inside and possibly reduces
# message size sent over the pipe
if record.args:
record.msg = record.msg % record.args
record.args = None
if record.exc_info:
dummy = self.format(record)
record.exc_info = None
return record
def emit(self, record):
try:
s = self._format_record(record)
self.send(s)
except (KeyboardInterrupt, SystemExit):
raise
except:
self.handleError(record)
def close(self):
self._handler.close()
logging.Handler.close(self)
concurrent-log-handler似乎完美地完成了这项工作。在Windows上测试。还支持POSIX系统。
主要思想
使用返回记录器的函数创建一个单独的文件。记录器必须为每个进程拥有ConcurrentRotatingFileHandler的新实例。示例函数get_logger()如下所示。 创建记录器是在流程初始化时完成的。对于多处理。进程的子类，它将意味着run()方法的开始。
详细说明
在这个例子中，我将使用下面的文件结构
.
│-- child.py <-- For a child process
│-- logs.py <-- For setting up the logs for the app
│-- main.py <-- For a main process
│-- myapp.py <-- For starting the app
│-- somemodule.py <-- For an example, a "3rd party module using standard logging"
Code
子进程
# child.py
import multiprocessing as mp
import time
from somemodule import do_something
class ChildProcess(mp.Process):
def __init__(self):
self.logger = None
super().__init__()
def run(self):
from logs import get_logger
self.logger = get_logger()
while True:
time.sleep(1)
self.logger.info("Child process")
do_something()
Simple child process that inherits multiprocessing.Process and simply logs to file text "Child process" Important: The get_logger() is called inside the run(), or elsewhere inside the child process (not module level or in __init__().) This is required as get_logger() creates ConcurrentRotatingFileHandler instance, and new instance is needed for each process. The do_something is used just to demonstrate that this works with 3rd party library code which does not have any clue that you are using concurrent-log-handler.
主要过程
# main.py
import logging
import multiprocessing as mp
import time
from child import ChildProcess
from somemodule import do_something
class MainProcess(mp.Process):
def __init__(self):
self.logger = logging.getLogger()
super().__init__()
def run(self):
from logs import get_logger
self.logger = get_logger()
self.child = ChildProcess()
self.child.daemon = True
self.child.start()
while True:
time.sleep(0.5)
self.logger.critical("Main process")
do_something()
主进程，在第二个“主进程”中两次登录到文件。同样继承自multiprocessing.Process。 get_logger()和do_something()的注释与子进程相同。
日志设置
# logs.py
import logging
import os
from concurrent_log_handler import ConcurrentRotatingFileHandler
LOGLEVEL = logging.DEBUG
def get_logger():
logger = logging.getLogger()
if logger.handlers:
return logger
# Use an absolute path to prevent file rotation trouble.
logfile = os.path.abspath("mylog.log")
logger.setLevel(LOGLEVEL)
# Rotate log after reaching 512K, keep 5 old copies.
filehandler = ConcurrentRotatingFileHandler(
logfile, mode="a", maxBytes=512 * 1024, backupCount=5, encoding="utf-8"
)
filehandler.setLevel(LOGLEVEL)
# create also handler for displaying output in the stdout
ch = logging.StreamHandler()
ch.setLevel(LOGLEVEL)
formatter = logging.Formatter(
"%(asctime)s - %(module)s - %(levelname)s - %(message)s [Process: %(process)d, %(filename)s:%(funcName)s(%(lineno)d)]"
)
# add formatter to ch
ch.setFormatter(formatter)
filehandler.setFormatter(formatter)
logger.addHandler(ch)
logger.addHandler(filehandler)
return logger
这使用了concurrent-log-handler包中的ConcurrentRotatingFileHandler。每个进程都需要一个新的ConcurrentRotatingFileHandler实例。 注意，ConcurrentRotatingFileHandler的所有参数在每个进程中都应该是相同的。
示例应用程序
# myapp.py
if __name__ == "__main__":
from main import MainProcess
p = MainProcess()
p.start()
这只是一个关于如何启动多进程应用程序的简单示例
第三方模块使用标准日志记录的例子
# somemodule.py
import logging
logger = logging.getLogger("somemodule")
def do_something():
logging.info("doing something")
只是一个简单的例子来测试来自第三方代码的记录器是否正常工作。
示例输出
2021-04-19 19:02:29,425 - main - CRITICAL - Main process [Process: 103348, main.py:run(23)]
2021-04-19 19:02:29,427 - somemodule - INFO - doing something [Process: 103348, somemodule.py:do_something(7)]
2021-04-19 19:02:29,929 - main - CRITICAL - Main process [Process: 103348, main.py:run(23)]
2021-04-19 19:02:29,931 - somemodule - INFO - doing something [Process: 103348, somemodule.py:do_something(7)]
2021-04-19 19:02:30,133 - child - INFO - Child process [Process: 76700, child.py:run(18)]
2021-04-19 19:02:30,137 - somemodule - INFO - doing something [Process: 76700, somemodule.py:do_something(7)]
2021-04-19 19:02:30,436 - main - CRITICAL - Main process [Process: 103348, main.py:run(23)]
2021-04-19 19:02:30,439 - somemodule - INFO - doing something [Process: 103348, somemodule.py:do_something(7)]
2021-04-19 19:02:30,944 - main - CRITICAL - Main process [Process: 103348, main.py:run(23)]
2021-04-19 19:02:30,946 - somemodule - INFO - doing something [Process: 103348, somemodule.py:do_something(7)]
2021-04-19 19:02:31,142 - child - INFO - Child process [Process: 76700, child.py:run(18)]
2021-04-19 19:02:31,145 - somemodule - INFO - doing something [Process: 76700, somemodule.py:do_something(7)]
2021-04-19 19:02:31,449 - main - CRITICAL - Main process [Process: 103348, main.py:run(23)]
2021-04-19 19:02:31,451 - somemodule - INFO - doing something [Process: 103348, somemodule.py:do_something(7)]
