Python多线程时序同步故障排查：A/C周期40ms，B处理耗时180ms

核心问题分析 #

新增线程B后出现停滞，大概率是以下逻辑错误导致：

• 生产消费速率不匹配引发阻塞：线程A每40ms生成1条数据（每秒25条），线程B单次处理需180ms（每秒约5.5条），A的生产速率远高于B的处理速率。若使用有界队列且未做阻塞处理，队列会快速被填满，导致A阻塞无法继续生产；若用无界队列，后续可能引发内存溢出，但前期表现为C长时间拿不到新数据。
• 线程B未持续循环处理：若B仅执行一次数据处理就退出，后续A生成的数据无人消费，C从空队列取数时会持续阻塞。
• 同步机制使用不当：比如条件变量未正确唤醒、锁未及时释放，导致某线程永久阻塞在等待状态。

修正后的实现代码（Python） #

import threading
import time
from queue import Queue

# 定义有界队列，平衡生产消费速率，避免内存溢出
data_queue_A_to_B = Queue(maxsize=10)
data_queue_B_to_C = Queue(maxsize=10)

def thread_A():
    """每40ms生成数据，放入A→B队列"""
    count = 0
    while True:
        data = f"数据_{count}"
        try:
            # 队列满时自动阻塞，等待B消费
            data_queue_A_to_B.put(data, block=True)
            print(f"[A] 生成: {data} | 时间: {time.time():.2f}")
            count += 1
            time.sleep(0.04)  # 40ms间隔
        except Exception as e:
            print(f"[A] 异常: {e}")
            break

def thread_B():
    """持续从A取数处理，耗时180ms后放入B→C队列"""
    while True:
        try:
            # 队列空时阻塞等待数据
            data = data_queue_A_to_B.get(block=True)
            # 模拟180ms处理耗时
            time.sleep(0.18)
            processed_data = f"处理后_{data}"
            data_queue_B_to_C.put(processed_data, block=True)
            print(f"[B] 处理完成: {processed_data} | 时间: {time.time():.2f}")
            # 标记任务完成，避免队列join阻塞
            data_queue_A_to_B.task_done()
        except Exception as e:
            print(f"[B] 异常: {e}")
            break

def thread_C():
    """每40ms从B取数展示，滞后A约180ms"""
    while True:
        try:
            # 带超时的阻塞取数，避免空队列永久阻塞
            data = data_queue_B_to_C.get(block=True, timeout=0.05)
            print(f"[C] 展示: {data} | 时间: {time.time():.2f}")
            data_queue_B_to_C.task_done()
        except:
            # 队列空时等待40ms再尝试，保证展示频率
            time.sleep(0.04)

if __name__ == "__main__":
    # 启动守护线程，主线程退出时自动终止
    t_a = threading.Thread(target=thread_A, daemon=True)
    t_b = threading.Thread(target=thread_B, daemon=True)
    t_c = threading.Thread(target=thread_C, daemon=True)
    
    t_a.start()
    t_b.start()
    t_c.start()
    
    # 主线程保持运行，捕获中断信号终止程序
    try:
        t_a.join()
        t_b.join()
        t_c.join()
    except KeyboardInterrupt:
        print("程序终止")

关键修正点 #

• 有界队列+阻塞处理：设置队列最大容量，生产线程（A）在队列满时自动阻塞，给B留出处理积压数据的时间，避免无限制生产导致内存问题。
• 线程B持续循环：用while True实现持续消费，确保A生成的所有数据都会被处理。
• C的取数逻辑优化：带超时的阻塞取数+空队列时的固定等待，既保证每40ms的展示频率，又不会因队列空而永久阻塞。
• 任务完成标记：调用task_done()，避免后续若需等待队列处理完成时出现阻塞。

时序验证 #

• A每40ms生成一条数据，B处理一条需180ms，队列会短暂积压，但A的阻塞机制会自动平衡速率；
• C每40ms尝试取数，由于B的处理耗时，展示的数据会滞后A约180ms，完全符合需求；
• 修正后三个线程会持续运行，不会出现停滞现象。

内容的提问来源于stack exchange，提问作者x xx