目前我在一个框架中有一个中央模块，使用Python 2.6的multiprocessing模块来生成多个进程。因为它使用了multiprocessing，所以有模块级别的多进程感知日志，LOG = multiprocessing.get_logger()。根据文档，这个记录器(编辑)没有进程共享锁，因此您不会通过让多个进程同时写入它来破坏sys.stderr(或任何文件句柄)。
我现在遇到的问题是，框架中的其他模块都不具备多进程感知能力。我认为，我需要使所有对这个中央模块的依赖都使用多进程感知日志记录。这在框架内部就很麻烦，更不用说对于框架的所有客户端了。我是否有没有考虑到的替代方案？
我刚刚编写了一个自己的日志处理程序，通过管道将所有内容传输到父进程。我只测试了十分钟，但它似乎工作得很好。
(注意：这是硬编码为RotatingFileHandler，适用于我的特定情况。)
现在使用队列来正确处理并发，并且能够正确地从错误中恢复。我已经在生产环境中使用了几个月，下面的当前版本可以正常工作。
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)
multiprocessing.Queue 在 put() 方法中使用了线程。因此在创建所有子进程之前不要调用 put(比如使用MultiProcessingLog处理器记录消息)。否则线程将会在子进程中死亡。一种解决方案是在每个子进程开始时调用 Queue._after_fork() 方法，或者改用 multiprocessing.queues.SimpleQueue，它不涉及线程但是是阻塞式的。 - Danqi Wang处理这个问题的唯一非侵入式方法是：
select，对可用的日志条目执行归并排序并刷新到中央日志。重复此过程。）multiprocessing.Queue，和/或者性能是一个问题，那么使用multiprocessing.Queue并不会更简单。 - vladrQueueHandler是Python 3.2+中的本地功能，可以完美实现此功能。在早期版本中也很容易复制。
Python文档提供了两个完整的示例：从多个进程记录到单个文件
每个进程（包括父进程）都将其日志放在Queue中，然后一个listener线程或进程（每个示例都提供了一个）会将它们全部取出并写入文件 - 没有损坏或混淆的风险。
对于使用Python < 3.2的用户，请导入logutils（与python 3.2本地代码相同）。
QueueHandler 诞生以来，这应该是被接受的答案。它不会对原有代码造成侵入式改动，透明并且能够工作，而且无论主进程使用什么样的日志配置，工作进程始终在记录到他们配置的 QueueHandler 中。此外，它也不需要父进程向派生的子进程提供任何日志器配置信息。 - g.pickardou以下是另一个专注于简单性的解决方案，适用于其他任何人（像我一样）从Google搜索到这里。记录应该很容易！仅适用于3.2或更高版本。
import multiprocessing
import logging
from logging.handlers import QueueHandler, QueueListener
import time
import random
def f(i):
time.sleep(random.uniform(.01, .05))
logging.info('function called with {} in worker thread.'.format(i))
time.sleep(random.uniform(.01, .05))
return i
def worker_init(q):
# all records from worker processes go to qh and then into q
qh = QueueHandler(q)
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
logger.addHandler(qh)
def logger_init():
q = multiprocessing.Queue()
# this is the handler for all log records
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(levelname)s: %(asctime)s - %(process)s - %(message)s"))
# ql gets records from the queue and sends them to the handler
ql = QueueListener(q, handler)
ql.start()
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
# add the handler to the logger so records from this process are handled
logger.addHandler(handler)
return ql, q
def main():
q_listener, q = logger_init()
logging.info('hello from main thread')
pool = multiprocessing.Pool(4, worker_init, [q])
for result in pool.map(f, range(10)):
pass
pool.close()
pool.join()
q_listener.stop()
if __name__ == '__main__':
main()
截至2020年，似乎有一种更简单的方式可以使用多进程进行日志记录。
此函数将创建记录器。您可以在此设置格式以及输出位置（文件、标准输出）：
def create_logger():
import multiprocessing, logging
logger = multiprocessing.get_logger()
logger.setLevel(logging.INFO)
formatter = logging.Formatter(\
'[%(asctime)s| %(levelname)s| %(processName)s] %(message)s')
handler = logging.FileHandler('logs/your_file_name.log')
handler.setFormatter(formatter)
# this bit will make sure you won't have
# duplicated messages in the output
if not len(logger.handlers):
logger.addHandler(handler)
return logger
if __name__ == '__main__':
from multiprocessing import Pool
logger = create_logger()
logger.info('Starting pooling')
p = Pool()
# rest of the code
logger = create_logger()
并输出消息：
logger.info(f'My message from {something}')
logging.info(..)。如果可以的话，我很乐意更新答案。 - Iopheamcreate_logger()，您可以在子进程内部设置 globals()['logger'] = multiprocessing.get_logger() 一次，假设您的函数使用 logger.info(...)。 - Jean Monet另一种选择可能是在logging包中使用各种非基于文件的日志处理程序：
SocketHandlerDatagramHandlerSyslogHandler（还有其他的）
这样，您可以轻松地拥有一个安全地写入并正确处理结果的日志守护进程。例如，一个简单的套接字服务器只需解压消息并将其发射到自己的旋转文件处理程序即可。
SyslogHandler也会为您处理此事。当然，您可以使用自己的syslog实例，而不是系统的实例。
"""sample code for logging in subprocesses using multiprocessing
* Little handler magic - The main process uses loggers and handlers as normal.
* Only a simple handler is needed in the subprocess that feeds the queue.
* Original logger name from subprocess is preserved when logged in main
process.
* As in the other implementations, a thread reads the queue and calls the
handlers. Except in this implementation, the thread is defined outside of a
handler, which makes the logger definitions simpler.
* Works with multiple handlers. If the logger in the main process defines
multiple handlers, they will all be fed records generated by the
subprocesses loggers.
tested with Python 2.5 and 2.6 on Linux and Windows
"""
import os
import sys
import time
import traceback
import multiprocessing, threading, logging, sys
DEFAULT_LEVEL = logging.DEBUG
formatter = logging.Formatter("%(levelname)s: %(asctime)s - %(name)s - %(process)s - %(message)s")
class SubProcessLogHandler(logging.Handler):
"""handler used by subprocesses
It simply puts items on a Queue for the main process to log.
"""
def __init__(self, queue):
logging.Handler.__init__(self)
self.queue = queue
def emit(self, record):
self.queue.put(record)
class LogQueueReader(threading.Thread):
"""thread to write subprocesses log records to main process log
This thread reads the records written by subprocesses and writes them to
the handlers defined in the main process's handlers.
"""
def __init__(self, queue):
threading.Thread.__init__(self)
self.queue = queue
self.daemon = True
def run(self):
"""read from the queue and write to the log handlers
The logging documentation says logging is thread safe, so there
shouldn't be contention between normal logging (from the main
process) and this thread.
Note that we're using the name of the original logger.
"""
# Thanks Mike for the error checking code.
while True:
try:
record = self.queue.get()
# get the logger for this record
logger = logging.getLogger(record.name)
logger.callHandlers(record)
except (KeyboardInterrupt, SystemExit):
raise
except EOFError:
break
except:
traceback.print_exc(file=sys.stderr)
class LoggingProcess(multiprocessing.Process):
def __init__(self, queue):
multiprocessing.Process.__init__(self)
self.queue = queue
def _setupLogger(self):
# create the logger to use.
logger = logging.getLogger('test.subprocess')
# The only handler desired is the SubProcessLogHandler. If any others
# exist, remove them. In this case, on Unix and Linux the StreamHandler
# will be inherited.
for handler in logger.handlers:
# just a check for my sanity
assert not isinstance(handler, SubProcessLogHandler)
logger.removeHandler(handler)
# add the handler
handler = SubProcessLogHandler(self.queue)
handler.setFormatter(formatter)
logger.addHandler(handler)
# On Windows, the level will not be inherited. Also, we could just
# set the level to log everything here and filter it in the main
# process handlers. For now, just set it from the global default.
logger.setLevel(DEFAULT_LEVEL)
self.logger = logger
def run(self):
self._setupLogger()
logger = self.logger
# and here goes the logging
p = multiprocessing.current_process()
logger.info('hello from process %s with pid %s' % (p.name, p.pid))
if __name__ == '__main__':
# queue used by the subprocess loggers
queue = multiprocessing.Queue()
# Just a normal logger
logger = logging.getLogger('test')
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(DEFAULT_LEVEL)
logger.info('hello from the main process')
# This thread will read from the subprocesses and write to the main log's
# handlers.
log_queue_reader = LogQueueReader(queue)
log_queue_reader.start()
# create the processes.
for i in range(10):
p = LoggingProcess(queue)
p.start()
# The way I read the multiprocessing warning about Queue, joining a
# process before it has finished feeding the Queue can cause a deadlock.
# Also, Queue.empty() is not realiable, so just make sure all processes
# are finished.
# active_children joins subprocesses when they're finished.
while multiprocessing.active_children():
time.sleep(.1)
fileConfig()，并在PoolWorkers中使用一个几乎没有配置的日志器（仅使用setLevel(logging.NOTSET)）。正如我在另一条评论中提到的那样，我正在使用Pool，因此我必须从Manager获取我的Queue（代理），而不是multiprocessing，以便它可以被pickled。这使我能够将队列传递给字典中的工作程序（其中大部分是使用vars()从argsparse对象派生的）。我觉得最终这是MS Windows的最佳方法，因为它缺少fork()并且打破了@zzzeak的解决方案。 - mltfork的MS Windows或其他平台上是行不通的。这样每个进程都会有自己独立的无用队列。链接的Q/A中的第二种方法在这些平台上也行不通。这是一种非可移植代码的方式。 - mltmultiprocessing.Queue，并且自那以后一直在使用它。虽然我不会声称理解为什么它能够正常工作。 - fantabolous当前所有解决方案都通过使用处理程序与日志配置过于耦合。我的解决方案具有以下架构和特点：
multiprocessing.Queue 与日志线程进行通信logging.Logger（以及已定义的实例）被修补以将所有记录发送到队列中代码示例和输出可在以下Gist中找到：https://gist.github.com/schlamar/7003737
daemon_thread.daemon设置为True。我需要这样做才能使我的Python程序在上下文管理器中发生异常时正确退出。 - blah238logged_call中捕获、记录和屏蔽目标func抛出的异常，否则异常会与其他记录的输出混淆。这是我修改后的版本：https://gist.github.com/blah238/8ab79c4fe9cdb254f5c37abfc5dc85bf - blah238daemon）能够出队和处理所有日志消息之前终止。您将失去这些消息。您如何在代码中处理此情况？ - Michele Piccolini由于我们可以将多进程日志记录表示为多个发布者和一个订阅者（监听器），因此使用ZeroMQ实现PUB-SUB消息传递确实是一种选择。
此外，PyZMQ模块，即Python绑定的ZMQ，实现了PUBHandler，它是用于通过zmq.PUB套接字发布日志消息的对象。
有一个网络解决方案，使用PyZMQ和PUBHandler从分布式应用程序进行集中日志记录，这可以很容易地采用，以便在本地使用多个发布进程工作。
formatters = {
logging.DEBUG: logging.Formatter("[%(name)s] %(message)s"),
logging.INFO: logging.Formatter("[%(name)s] %(message)s"),
logging.WARN: logging.Formatter("[%(name)s] %(message)s"),
logging.ERROR: logging.Formatter("[%(name)s] %(message)s"),
logging.CRITICAL: logging.Formatter("[%(name)s] %(message)s")
}
# This one will be used by publishing processes
class PUBLogger:
def __init__(self, host, port=config.PUBSUB_LOGGER_PORT):
self._logger = logging.getLogger(__name__)
self._logger.setLevel(logging.DEBUG)
self.ctx = zmq.Context()
self.pub = self.ctx.socket(zmq.PUB)
self.pub.connect('tcp://{0}:{1}'.format(socket.gethostbyname(host), port))
self._handler = PUBHandler(self.pub)
self._handler.formatters = formatters
self._logger.addHandler(self._handler)
@property
def logger(self):
return self._logger
# This one will be used by listener process
class SUBLogger:
def __init__(self, ip, output_dir="", port=config.PUBSUB_LOGGER_PORT):
self.output_dir = output_dir
self._logger = logging.getLogger()
self._logger.setLevel(logging.DEBUG)
self.ctx = zmq.Context()
self._sub = self.ctx.socket(zmq.SUB)
self._sub.bind('tcp://*:{1}'.format(ip, port))
self._sub.setsockopt(zmq.SUBSCRIBE, "")
handler = handlers.RotatingFileHandler(os.path.join(output_dir, "client_debug.log"), "w", 100 * 1024 * 1024, 10)
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s;%(levelname)s - %(message)s")
handler.setFormatter(formatter)
self._logger.addHandler(handler)
@property
def sub(self):
return self._sub
@property
def logger(self):
return self._logger
# And that's the way we actually run things:
# Listener process will forever listen on SUB socket for incoming messages
def run_sub_logger(ip, event):
sub_logger = SUBLogger(ip)
while not event.is_set():
try:
topic, message = sub_logger.sub.recv_multipart(flags=zmq.NOBLOCK)
log_msg = getattr(logging, topic.lower())
log_msg(message)
except zmq.ZMQError as zmq_error:
if zmq_error.errno == zmq.EAGAIN:
pass
# Publisher processes loggers should be initialized as follows:
class Publisher:
def __init__(self, stop_event, proc_id):
self.stop_event = stop_event
self.proc_id = proc_id
self._logger = pub_logger.PUBLogger('127.0.0.1').logger
def run(self):
self._logger.info("{0} - Sending message".format(proc_id))
def run_worker(event, proc_id):
worker = Publisher(event, proc_id)
worker.run()
# Starting subscriber process so we won't loose publisher's messages
sub_logger_process = Process(target=run_sub_logger,
args=('127.0.0.1'), stop_event,))
sub_logger_process.start()
#Starting publisher processes
for i in range(MAX_WORKERS_PER_CLIENT):
processes.append(Process(target=run_worker,
args=(stop_event, i,)))
for p in processes:
p.start()
import multiprocessing, logging, sys, re, os, StringIO, threading, time, Queue
class MultiProcessingLogHandler(logging.Handler):
def __init__(self, handler, queue, child=False):
logging.Handler.__init__(self)
self._handler = handler
self.queue = queue
# we only want one of the loggers to be pulling from the queue.
# If there is a way to do this without needing to be passed this
# information, that would be great!
if child == False:
self.shutdown = False
self.polltime = 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):
#print "receive on"
while (self.shutdown == False) or (self.queue.empty() == False):
# so we block for a short period of time so that we can
# check for the shutdown cases.
try:
record = self.queue.get(True, self.polltime)
self._handler.emit(record)
except Queue.Empty, e:
pass
def send(self, s):
# send just puts it in the queue for the server to retrieve
self.queue.put(s)
def _format_record(self, record):
ei = record.exc_info
if ei:
dummy = self.format(record) # just to get traceback text into record.exc_text
record.exc_info = None # to avoid Unpickleable error
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):
time.sleep(self.polltime+1) # give some time for messages to enter the queue.
self.shutdown = True
time.sleep(self.polltime+1) # give some time for the server to time out and see the shutdown
def __del__(self):
self.close() # hopefully this aids in orderly shutdown when things are going poorly.
def f(x):
# just a logging command...
logging.critical('function number: ' + str(x))
# to make some calls take longer than others, so the output is "jumbled" as real MP programs are.
time.sleep(x % 3)
def initPool(queue, level):
"""
This causes the logging module to be initialized with the necessary info
in pool threads to work correctly.
"""
logging.getLogger('').addHandler(MultiProcessingLogHandler(logging.StreamHandler(), queue, child=True))
logging.getLogger('').setLevel(level)
if __name__ == '__main__':
stream = StringIO.StringIO()
logQueue = multiprocessing.Queue(100)
handler= MultiProcessingLogHandler(logging.StreamHandler(stream), logQueue)
logging.getLogger('').addHandler(handler)
logging.getLogger('').setLevel(logging.DEBUG)
logging.debug('starting main')
# when bulding the pool on a Windows machine we also have to init the logger in all the instances with the queue and the level of logging.
pool = multiprocessing.Pool(processes=10, initializer=initPool, initargs=[logQueue, logging.getLogger('').getEffectiveLevel()] ) # start worker processes
pool.map(f, range(0,50))
pool.close()
logging.debug('done')
logging.shutdown()
print "stream output is:"
print stream.getvalue()
if 'MainProcess' == multiprocessing.current_process().name:来代替传递child？ - mlt
multiprocessing.get_logger()的用例是什么？看起来，基于这些其他记录日志的方式，multiprocessing中的记录日志功能很少有价值。 - Tim Ludwinskiget_logger()是multiprocessing模块本身使用的记录器。如果您想要调试一个multiprocessing问题，它会很有用。 - jfs