Python如何实现100秒内交替打印两组日志？

Question

在多线程日志系统中，如何使用Python在100秒内精确交替打印两组日志（如“Log A”和“Log B”），每组间隔1秒？常见问题是线程竞争导致输出顺序混乱或延迟累积。使用`threading.Timer`或`time.sleep()`时，容易因GIL或调度延迟破坏交替节奏。如何通过线程同步机制（如`Lock`或`Condition`）确保严格交替且总时长控制在100秒内？

Answer 1

多线程日志系统中精确交替打印日志的实现与优化

1. 问题背景与挑战分析

在高并发系统中，日志记录是调试、监控和审计的重要手段。当多个线程同时写入日志时，若缺乏同步机制，极易出现输出顺序混乱、时间漂移等问题。

以“Log A”和“Log B”两组日志每秒交替打印为例，理想情况应为：

1. 第0秒：Log A
2. 第1秒：Log B
3. 第2秒：Log A
4. 第3秒：Log B
5. ……
6. 第99秒：Log B（假设从A开始）

然而，使用time.sleep()或threading.Timer常因以下原因导致节奏失准：

• GIL调度延迟：Python的全局解释器锁可能导致线程切换不及时。
• 系统调度抖动：操作系统线程调度并非实时，微秒级延迟累积成显著偏差。
• 竞争条件：多个线程争抢资源，造成输出顺序不可控。

2. 常见解决方案对比

方法	精度	同步能力	延迟累积风险	适用场景
time.sleep()	低	弱	高	简单脚本
threading.Timer	中	弱	中	定时任务
Lock + sleep	中	强	中	基础同步
Condition + wait/notify	高	强	低	严格交替控制
queue.Queue + 主控线程	极高	极强	无	生产级日志系统

3. 基于 Condition 的严格交替实现

使用threading.Condition可实现线程间精确协调。核心思想是通过条件变量控制执行权的传递。

import threading
import time

class AlternatingLogger:
    def __init__(self):
        self.condition = threading.Condition()
        self.turn = 'A'  # 初始轮到A
        self.stop_flag = False
        self.start_time = time.time()

    def log_a(self):
        for _ in range(50):  # 总共100秒，每2秒一轮，共50轮
            with self.condition:
                while self.turn != 'A' and not self.stop_flag:
                    self.condition.wait()
                if self.stop_flag:
                    break
                print(f"{int(time.time() - self.start_time)}: Log A")
                self.turn = 'B'
                self.condition.notify_all()
                time.sleep(1)  # 固定间隔

    def log_b(self):
        for _ in range(50):
            with self.condition:
                while self.turn != 'B' and not self.stop_flag:
                    self.condition.wait()
                if self.stop_flag:
                    break
                print(f"{int(time.time() - self.start_time)}: Log B")
                self.turn = 'A'
                self.condition.notify_all()
                time.sleep(1)

    def start(self):
        self.start_time = time.time()
        ta = threading.Thread(target=self.log_a)
        tb = threading.Thread(target=self.log_b)
        ta.start()
        tb.start()
        ta.join()
        tb.join()

# 使用示例
logger = AlternatingLogger()
logger.start()

4. 高精度替代方案：事件驱动队列模型

为避免sleep带来的累积误差，可采用主控线程按时间表调度任务。以下使用queue.Queue与计时器结合：

import queue
import threading
import time

def event_driven_logger():
    q = queue.Queue()
    start_time = time.time()

    def producer():
        for i in range(100):
            target_time = start_time + i
            log_msg = "Log A" if i % 2 == 0 else "Log B"
            q.put((target_time, log_msg))
            time.sleep(1)  # 控制生成频率

    def consumer():
        while True:
            try:
                target_time, msg = q.get(timeout=1)
                delay = target_time - time.time()
                if delay > 0:
                    time.sleep(delay)
                print(f"{int(time.time() - start_time)}: {msg}")
                q.task_done()
            except queue.Empty:
                break

    t1 = threading.Thread(target=producer)
    t2 = threading.Thread(target=consumer)
    t1.start(); t2.start()
    t1.join(); t2.join()

5. 架构设计流程图

graph TD A[启动主程序] --> B[初始化Condition或Queue] B --> C[创建Log A线程] C --> D[创建Log B线程] D --> E[线程A获取锁] E --> F{是否轮到A?} F -- 是 --> G[打印Log A] F -- 否 --> H[等待通知] G --> I[设置轮次为B] I --> J[通知所有线程] J --> K[睡眠1秒] K --> L{完成50次?} L -- 否 --> E L -- 是 --> M[退出] H --> F

6. 性能调优与GIL影响规避

由于CPython的GIL限制，纯CPU密集型线程难以并行。但在I/O操作（如print）期间，GIL会释放，因此日志系统仍可受益于多线程。

优化建议：

• 减少临界区代码量，仅将共享状态访问置于锁内。
• 使用logging模块而非直接print，因其内部已做线程安全处理。
• 考虑异步IO（asyncio）替代多线程，在单线程中调度日志任务，避免GIL竞争。
• 启用线程优先级（需操作系统支持）提升调度确定性。