Python-threading:单独记录各个线程

Python-threading: separate logging of individual threads

我希望每个线程都有单独的 .log 文件。不幸的是,在使用 logging.basicConfig 之后,为日志创建了许多不同的文件,但最终所有日志都在最后声明的文件中结束。 线程应该怎么做才能拥有独立的日志文件?

import logging
import threading
import time
from datetime import datetime

def test_printing(name):
    logging.basicConfig(
        format="%(asctime)s, %(levelname)-8s | %(filename)-23s:%(lineno)-4s | %(threadName)15s: %(message)s",  # noqa
        datefmt="%Y-%m-%d:%H:%M:%S",
        level=logging.INFO,
        force=True,
        handlers=[
            logging.FileHandler(f"{name}.log"),
            logging.StreamHandler()])
    logging.info(f"Test {name}")
    time.sleep(20)
    logging.info(f"Test {name} after 20s")


def function_thread():
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    thread = threading.Thread(
            target=test_printing,
            kwargs={"name": timestamp}
        )
    thread.start()

for i in range(5):
    time.sleep(1)
    function_thread()

来自https://docs.python.org/3/library/logging.html#logger-objects

Note that Loggers should NEVER be instantiated directly, but always through the module-level function logging.getLogger(name).

所以你必须在每个线程中创建和配置一个新的记录器:

logger = logging.getLogger(name)
logger.basicConfig(...)

更多信息:https://docs.python.org/3/howto/logging.html#logging-from-multiple-modules

编辑:使用已经定义的 name 作为记录器标识符,而不是 __name__


编辑: 您不能使用 logging.basicConfig,而是需要单独配置每个线程记录器。 提供和测试的完整代码:

import logging
import threading
import time
from datetime import datetime

def test_printing(name):
    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)
    formatter = logging.Formatter(
        fmt="%(asctime)s, %(levelname)-8s | %(filename)-23s:%(lineno)-4s | %(threadName)15s: %(message)s",
        datefmt="%Y-%m-%d:%H:%M:%S")
    sh = logging.StreamHandler()
    fh = logging.FileHandler(f"{name}.log")
    sh.setFormatter(formatter)
    fh.setFormatter(formatter)
    logger.addHandler(sh)
    logger.addHandler(fh)

    logger.info(f"Test {name}")
    time.sleep(20)
    logger.info(f"Test {name} after 20s")


def function_thread():
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    thread = threading.Thread(
            target=test_printing,
            kwargs={"name": timestamp}
        )
    thread.start()


for i in range(5):
    time.sleep(1)
    function_thread()