Python线程锁竞争场景下OS线程调度机制与锁获取规律的技术问询

API 探路者

2026-4-15

我先明确下，这个问题和另一个关于Python锁总是被同一个线程重入的问题不一样。表面上看两者都用了两个线程，但那个问题是要求每个线程执行一次再切换，实现交替打印的特定模式，核心是找实现该目标的具体方案。而我的问题更偏向原理探究：我不需要特定的执行模式，而是想搞懂我观察到的现象背后的逻辑，以及哪些变量会影响这个现象，方便我进一步研究。

import threading
import time

lock = threading.Lock()


def high_priority():
    while True:
        with lock:
            print("High priority thread acquired lock")
            time.sleep(0.2)


def low_priority():
    while True:
        with lock:
            print("Low priority thread acquired lock")  # May never happen!
            time.sleep(0.1)


t1 = threading.Thread(target=high_priority, daemon=True)
t2 = threading.Thread(target=low_priority, daemon=True)

t1.start()
t2.start()

time.sleep(1)

我运行这段代码23次，才第一次看到低优先级线程拿到锁。

我的核心目标是搞清楚：

在主线程sleep的1秒内，低优先级线程什么时候能拿到锁
低优先级线程拿到锁的频率由什么决定

不过直接做实验太耗时，所以想先搞懂背后的理论。

下面是23次运行里最后3次的输出：

➜  deadlock python deadlock_priority.py
High priority thread acquired lock
High priority thread acquired lock
High priority thread acquired lock
High priority thread acquired lock
High priority thread acquired lock
➜  deadlock python deadlock_priority.py
High priority thread acquired lock
High priority thread acquired lock
High priority thread acquired lock
High priority thread acquired lock
High priority thread acquired lock
➜  deadlock python deadlock_priority.py
High priority thread acquired lock
High priority thread acquired lock
High priority thread acquired lock
Low priority thread acquired lock
Low priority thread acquired lock
Low priority thread acquired lock
Low priority thread acquired lock

我还有一堆疑问想请教：

线程的“优先级”是由启动顺序（t1.start()在前，t2.start()在后）决定的吗？还是由线程内部sleep的时长决定？
我本来想通过缩短低优先级线程内部的sleep时间来提高它拿到锁的概率，这是不是一厢情愿？我怀疑锁上下文里的time.sleep设置其实没什么影响？
有没有办法让这个实验的结果更可复现？比如能保证在固定次数的运行里，低优先级线程一定能拿到锁？
还有哪些参数可以调整或者添加到代码里，能帮助我更好地理解这类线程竞争问题？
我写这段代码只是为了研究线程竞争，没有实际的应用场景，但直觉告诉我这种模式在实际代码里应该有用。有没有人能解释下，这类线程竞争的场景会出现在哪些有实际意义的代码中？

后来我试着在锁的上下文外面随机加了sleep，结果低优先级线程拿到锁的频率一下子高了很多：

def high_priority():
    while True:
        time.sleep(0.0001)
        with lock:
            print("High priority thread acquired lock")
            time.sleep(0.2)


def low_priority():
    while True:
        time.sleep(0.0001)
        with lock:
            print("Low priority thread acquired lock")
            time.sleep(0.1)

修改后前3次运行的输出：

➜  deadlock python deadlock_sleep.py 
High priority thread acquired lock
Low priority thread acquired lock
High priority thread acquired lock
Low priority thread acquired lock
High priority thread acquired lock
Low priority thread acquired lock
High priority thread acquired lock
➜  deadlock python deadlock_sleep.py
High priority thread acquired lock
Low priority thread acquired lock
High priority thread acquired lock
Low priority thread acquired lock
High priority thread acquired lock
Low priority thread acquired lock
High priority thread acquired lock
➜  deadlock python deadlock_sleep.py
High priority thread acquired lock
Low priority thread acquired lock
High priority thread acquired lock
Low priority thread acquired lock
High priority thread acquired lock
Low priority thread acquired lock
High priority thread acquired lock

备注：内容来源于stack exchange，提问作者Han Qi