Python INFO 到 CONSOLE 和 DEBUG 到 FILE 日志记录

Question

所以我几天来一直在努力寻找解决问题的方法......

我有以下结构

main.py
   -> module1.py
     -> task (Threadable Class)
   -> module2.py
     -> task (Threadable Class)

在我的主目录中，用户可以选择一个模块以及在该模块中启动的任务数量。每次发生新事情时，我都会打印（）他们的状态。来自 Web 服务器的有效响应示例。但是这样做会导致输出变得混乱，因为多个线程同时调用 print。即使 flush=True 在这里也没有帮助。

据我所知，实现良好输出的唯一解决方案是使用日志记录。日志记录已经实现，因为程序在启动时创建一个带有调试信息的日志，并在运行时不断写入。这是为了更好的错误跟踪和其他东西。

我尝试过以下独奏：

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(f'log/{CURR_LOGFILE}.txt'),
        logging.StreamHandler()
    ]
)

但这对我没有帮助，因为我希望将 logging.info() 打印到控制台，同时将 logging.debug() 打印到日志文件。

我将如何继续使这项工作和线程安全？或者是否有更好的方法让每个线程发出它们的状态？

锁定不起作用，因为它会临时锁定线程，并且在此用例中每毫秒都很重要。

Answer 1

从技术上讲，使用logging 并不是唯一的解决方案。您正确地确定了问题：多个线程共享同一个标准输出缓冲区，它们在其中同时写入。标准日志库恰好是 thread-safe，但没有它你也可以这样做¹。

您可以创建一个threading.Lock 并重写您的一些代码，以便线程在持有锁时只能print。
您的工作负载似乎并不繁重，因此性能应该不是问题。

---

¹ 我认为这是XY problem 的情况：您实际上并不想使用logging，如果我理解正确，您真正想要的只是同步您的输出，所以最后的显示没有乱七八糟。

Answer 2

使用日志记录是有效的，因为它在内部也使用了锁。这么说

我已经尝试过使用threading.Lock，但是由于必须在不停止运行的情况下执行代码，因此 100 个线程会导致其他线程必须等待直到它们持有锁

是错误的。

这是一个简单的基准，我并不打算让它防弹，只是为了向您展示两种方法的工作方式相似：

import logging
import time
import threading
import random
import sys


logger = logging.getLogger("so70643825")
logger.addHandler(logging.StreamHandler(sys.stdout))
logger.setLevel(logging.INFO)


lock = threading.Lock()
def lock_print(*args, **kwargs):
    with lock:
        print(*args, **kwargs)


def task_creator(task_number: int, display_function):
    def actual_task(x: int):
        for i in range(x):
            display_function(f"Task {task_number}, {i=}")
            time.sleep(random.random() / 10)  # sleep up to 0.1s
    return actual_task


def run_tasks(number_of_threads: int, number_of_iterations: int, display_function):
    threads = []
    for thread_number in range(number_of_threads):
        thread = threading.Thread(target=task_creator(task_number=thread_number, display_function=display_function),
                                  args=(number_of_iterations,))
        threads.append(thread)
        thread.start()
    for thread in threads:
        thread.join()


def main():
    N_THREADS = 100
    N_LOOPS = 10
    N_MEASURES = 10
    times = []

    for i in range(N_MEASURES):
        time_before_logging_tasks = time.time()
        run_tasks(N_THREADS, N_LOOPS, logger.info)
        time_after_logging_tasks = time.time()

        time_before_locking_tasks = time.time()
        run_tasks(N_THREADS, N_LOOPS, lock_print)
        time_after_locking_tasks = time.time()

        time_logging = time_after_logging_tasks - time_before_logging_tasks
        time_locking = time_after_locking_tasks - time_before_locking_tasks
        times.append((time_logging, time_locking,))

    for time_logging, time_locking in times:
        print(f"logging: {time_logging}   locking: {time_locking} best={'LOG' if time_logging < time_locking else 'LOCK'}")


if __name__ == "__main__":
    main()

结果：

logging: 0.7429051399230957   locking: 0.7113175392150879 best=LOCK
logging: 0.7250735759735107   locking: 0.6937775611877441 best=LOCK
logging: 0.712653636932373    locking: 0.691455602645874  best=LOCK
logging: 0.7124729156494141   locking: 0.6672348976135254 best=LOCK
logging: 0.7507524490356445   locking: 0.7678782939910889 best=LOG
logging: 0.690934419631958    locking: 0.7111172676086426 best=LOG
logging: 0.7271988391876221   locking: 0.7642250061035156 best=LOG
logging: 0.7295012474060059   locking: 0.7794029712677002 best=LOG
logging: 0.7717809677124023   locking: 0.7706160545349121 best=LOCK
logging: 0.7927803993225098   locking: 0.7397556304931641 best=LOCK

LOG 和 LOCK 混合获胜，但每次它们之间的差异很小。
删除随机的sleep 使 LOCK 成为明显的赢家（至少快十倍）。我认为这是因为睡眠使所有日志记录机器引入的开销相形见绌，而 lock_print 不需要。

如果性能对问题非常（我是认真的）很重要，那么你应该在最初就说出来。