快捷導(dǎo)航

python?logging多進(jìn)程多線(xiàn)程輸出到同一個(gè)日志文件的實(shí)戰(zhàn)案例

更新時(shí)間：2022年02月22日 14:41:17 作者：VincentZhu

這篇文章主要介紹了python?logging多進(jìn)程多線(xiàn)程輸出到同一個(gè)日志文件的實(shí)戰(zhàn)案例，本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下

參考官方案例：https://docs.python.org/zh-cn/3.8/howto/logging-cookbook.html

import logging
import logging.config
import logging.handlers
from multiprocessing import Process, Queue
import random
import threading
import time

def logger_thread(q):
    while True:
        record = q.get()
        if record is None:
            break
        logger = logging.getLogger(record.name)
        logger.handle(record)
def worker_process(q):
    qh = logging.handlers.QueueHandler(q)
    root = logging.getLogger()
    root.setLevel(logging.DEBUG)
    root.addHandler(qh)
    levels = [logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR,
              logging.CRITICAL]
    loggers = ['foo', 'foo.bar', 'foo.bar.baz',
               'spam', 'spam.ham', 'spam.ham.eggs']
    for i in range(100):
        lvl = random.choice(levels)
        logger = logging.getLogger(random.choice(loggers))
        logger.log(lvl, 'Message no. %d', i)
if __name__ == '__main__':
    q = Queue()
    d = {
        'version': 1,
        'formatters': {
            'detailed': {
                'class': 'logging.Formatter',
                'format': '%(asctime)s %(name)-15s %(levelname)-8s %(processName)-10s %(message)s'
            }
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO',
            },
            'file': {
                'class': 'logging.FileHandler',
                'filename': 'mplog.log',
                'mode': 'w',
                'formatter': 'detailed',
            'foofile': {
                'filename': 'mplog-foo.log',
            'errors': {
                'filename': 'mplog-errors.log',
                'level': 'ERROR',
        'loggers': {
            'foo': {
                'handlers': ['foofile']
        'root': {
            'level': 'DEBUG',
            'handlers': ['console', 'file', 'errors']
    }
    workers = []
    for i in range(5):
        wp = Process(target=worker_process, name='worker %d' % (i + 1), args=(q,))
        workers.append(wp)
        wp.start()
    logging.config.dictConfig(d)
    lp = threading.Thread(target=logger_thread, args=(q,))
    lp.start()
    # At this point, the main process could do some useful work of its own
    # Once it's done that, it can wait for the workers to terminate...
    for wp in workers:
        wp.join()
    # And now tell the logging thread to finish up, too
    q.put(None)
    lp.join()

實(shí)戰(zhàn)案例：

1、字典形式配置日志

log_conf_dict = {
    'version': 1,
    'formatters': {
        'my_formatter': {
            'class': 'logging.Formatter',
            'format': '%(asctime)s %(processName)s(%(process)d) %(threadName)s(%(thread)d) %(filename)s[line:%(lineno)d] %(levelname)s %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'my_formatter',
        },
        'file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'filename': '/log/test.log',
            'maxBytes': 5*1024*1024,
            'backupCount': 60,
            'mode': 'w',
            'delay': True,
            'formatter': 'my_formatter',
            'encoding': 'utf-8',
            'level': 'INFO',
        },
    },
    'loggers': {
        'my_logger': {
            'handlers': ['file']
        }
    },
    'root': {
        'level': _level,
        'handlers': ['console', 'file']
    },
}

2、主進(jìn)程中開(kāi)啟獨(dú)立的日志寫(xiě)入監(jiān)聽(tīng)線(xiàn)程

"""主進(jìn)程中開(kāi)啟獨(dú)立的日志寫(xiě)入監(jiān)聽(tīng)線(xiàn)程"""
    queue = Queue(-1)
    logging.config.dictConfig(dict)
    log_thread = threading.Thread(target=logger_main, args=(queue,))
    log_thread.start()    """其他邏輯代碼段"""    queue.put(None)    log_thread.join()

日志寫(xiě)入函數(shù)

def logger_main(q):
'''日志隊(duì)列寫(xiě)入文件'''
    while True:
        record = q.get()
        if record is None:
            break
        logger = logging.getLogger()
        logger.handle(record)

3、子進(jìn)程中將日志輸入QueueHandler日志隊(duì)列

def child_proc_main(queue):
    lqh = logging.handlers.QueueHandler(queue)
    lqh.set_name("my_queue_handler")
    
    root = logging.getLogger()    #很關(guān)鍵的一步，必須先清空，再加入。原因：多進(jìn)程多線(xiàn)程復(fù)雜環(huán)境下，在window和linux平臺(tái)運(yùn)行表現(xiàn)不一致，linux會(huì)復(fù)制主進(jìn)程的日志配置，造成同時(shí)輸出多個(gè)日志文件。
    root.handlers.clear()
    root.addHandler(lqh)
    root.setLevel(level)

到此這篇關(guān)于python logging多進(jìn)程多線(xiàn)程輸出到同一個(gè)日志文件的文章就介紹到這了,更多相關(guān)python logging日志文件內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: