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

Python主线程执行时如何锁定类属性避免线程数据竞争问题

线程安全问题:数据更新与API调用的冲突

我主脚本启动了一个API,同时用另一个线程执行每日数据更新任务。目前每日更新会在API调用过程中修改数据,从而引发错误。我将该问题简化为如下代码示例:

import schedule
from threading import Thread, Lock
import time


def periodically_run_pending_tasks(interval=0.5):
    while True:
        schedule.run_pending()
        time.sleep(interval)


def change_state(item):
    item.a = "New state"


class MyClass:
    def __init__(self):
        self.a = "Initial state"

    def print_state_twice(self, lk):
        lk.acquire()
        print(self.a)
        time.sleep(2)
        print(self.a)
        lk.release()


my_class = MyClass()
lock = Lock()

thread = Thread(target=periodically_run_pending_tasks)
schedule.every().second.do(change_state, my_class)
thread.start()

my_class.print_state_twice(lock)

我期望该脚本输出两次Initial state或两次New state,但实际返回Initial state和New state。我还尝试了另一种写法:

import schedule
from threading import Thread, Lock
import time


def periodically_run_pending_tasks(interval=0.5):
    while True:
        schedule.run_pending()
        time.sleep(interval)


def change_state(item):
    item.a = "New state"


class MyClass:
    def __init__(self):
        self.a = "Initial state"

    def print_state_twice(self, lk):
        lk.acquire()
        print(self.a)
        time.sleep(2)
        print(self.a)
        lk.release()


my_class = MyClass()
lock = Lock()

thread_2 = Thread(target=my_class.print_state_twice, args=(lock,))
thread_2.start()

thread = Thread(target=periodically_run_pending_tasks)
schedule.every().second.do(change_state, my_class)
thread.start()

内容的提问来源于stack exchange,提问作者JulienBr

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.08.05 22:55:37