You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Python中实现多线程交替执行?

实现两个线程交替执行的几种方法

你的代码里两个线程独立运行,没有同步机制,操作系统的线程调度可能让其中一个线程先完成所有打印操作。要实现交替输出,需要通过线程同步工具来协调两个线程的执行顺序,以下是几种可行方案:

方案一:使用两个互斥锁(threading.Lock)

通过两个锁的交替获取与释放,强制线程按顺序执行:

import threading

lock1 = threading.Lock()
lock2 = threading.Lock()

# 先锁住lock2,让test2初始无法执行
lock2.acquire()

def test1():
    for _ in range(3):
        lock1.acquire()
        print("Test1")
        lock2.release()

def test2():
    for _ in range(3):
        lock2.acquire()
        print("Test2")
        lock1.release()

t1 = threading.Thread(target=test1)
t2 = threading.Thread(target=test2)

t1.start()
t2.start()

t1.join()
t2.join()

原理:初始时lock2处于锁定状态,test2会卡在lock2.acquire();test1可以获取lock1,打印后释放lock2让test2执行;test2打印后释放lock1,让test1进入下一次循环,以此实现交替。

方案二:使用条件变量(threading.Condition)

通过条件变量控制线程的等待与唤醒,配合共享标记位判断执行权限:

import threading

cond = threading.Condition()
current_executor = "test1"  # 标记当前可执行的线程

def test1():
    global current_executor
    for _ in range(3):
        with cond:
            # 等待直到轮到自己执行
            while current_executor != "test1":
                cond.wait()
            print("Test1")
            current_executor = "test2"
            cond.notify()  # 唤醒test2线程

def test2():
    global current_executor
    for _ in range(3):
        with cond:
            while current_executor != "test2":
                cond.wait()
            print("Test2")
            current_executor = "test1"
            cond.notify()

t1 = threading.Thread(target=test1)
t2 = threading.Thread(target=test2)

t1.start()
t2.start()

t1.join()
t2.join()

原理:with cond会自动处理锁的获取与释放。线程进入后检查是否拥有执行权限,没有则进入等待状态;执行完成后切换标记位,唤醒另一个线程,确保交替执行。

方案三:使用事件对象(threading.Event)

通过两个事件的交替触发,控制线程的执行时机:

import threading

event1 = threading.Event()
event2 = threading.Event()

event1.set()  # 初始让test1可以执行

def test1():
    for _ in range(3):
        event1.wait()
        print("Test1")
        event1.clear()
        event2.set()

def test2():
    for _ in range(3):
        event2.wait()
        print("Test2")
        event2.clear()
        event1.set()

t1 = threading.Thread(target=test1)
t2 = threading.Thread(target=test2)

t1.start()
t2.start()

t1.join()
t2.join()

原理:初始设置event1,test1等待事件触发后打印,随后清除event1并触发event2;test2等待event2触发后打印,再清除event2触发event1,循环实现交替输出。

内容的提问来源于stack exchange,提问作者Super Kai - Kazuya Ito

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.10 21:25:38