Python threading Local()函数用法案例详解

苏见山

发布时间：2026-10-13| 3人浏览过

原创

本文详解Python threading模块中local()函数的用法。通过多线程访问公共资源导致数据竞争的案例，展示如何利用local()实现线程间数据隔离。文章对比了全局变量与ThreadLocal对象的行为，并深入剖析其底层实现原理，帮助开发者理解如何避免线程安全问题。

前言

在多线程环境中，若多个线程共同修改同一公共资源，极易因数据不同步而引发线程安全问题。通常，我们采用互斥锁来保护公共资源以解决此类问题。

然而，若资源无需在多线程间共享，则不必强制设为公共资源。此时，可利用Python threading模块提供的local()机制来规避线程安全问题。该机制与Java中的ThreadLocal类功能相似，感兴趣者可参考Java ThreadLocal的原理解析与应用案例。

local() 函数是什么？

threading模块中的local()函数旨在封装公共资源，使同一资源在不同线程间保持隔离。为便于理解，可将其比喻为一个公共大箱子（代表公共资源），每个线程（用户）将手机放入箱中。若无控制，取手机时易拿错；而使用local()则如同对箱子进行标记管理，每个手机存入时标记所有者并隔离存放，确保取出时能准确找到属于自己的手机。

调用local()函数会生成一个ThreadLocal对象，该对象对所有线程可见，如同上述的大箱子。但存入该对象的变量却是各线程独有的，即使变量名相同，其指向的值也完全独立。

local()函数如何用？

local()函数的基本使用语法如下：

import threading

local=threading.local()

首先引入threading模块，随后调用local()函数获取全局的ThreadLocal对象。为使说明更生动，我们继续以“大箱子”为例进行代码演示。

1. 不做标记，不做隔离

第一个示例演示所有线程将手机放入大箱子，既不标记也不隔离。先存入，延时后再取出。

import threading
import time


def set_telephone(telephone):
    global global_telephone
    global_telephone = telephone
    print(threading.current_thread().name + " 放入的手机是", global_telephone)
    time.sleep(1)
    get_telephone()


def get_telephone():
    print(threading.current_thread().name + " 取出的手机是", global_telephone)


if __name__ == '__main__':
    for i in range(3):
        thread = threading.Thread(target=set_telephone, name='学生' + str(i), args=('手机' + str(i),))
        thread.start()

运行结果如下：

学生0 放入的手机是手机0

学生1 放入的手机是手机1

学生2 放入的手机是手机2

学生0 取出的手机是手机2

学生1 取出的手机是手机2

学生2 取出的手机是手机2

此处有三个线程，分别模拟学生0、学生1和学生2，将各自手机赋值给全局变量global_telephone（即大箱子），随后读取该全局变量的值。结果显示所有线程取出的均为手机2，这显然不符合预期，体现了缺乏控制机制的后果。

2.使用local()函数加以控制

使用local()函数进行控制时，需将全局变量替换为ThreadLocal对象，由该对象管理各线程中的值。

import threading
import time


def set_telephone(telephone):
    local.telephone = telephone
    print(threading.current_thread().name + " 放入的手机是", local.telephone + "\n")
    time.sleep(1)
    get_telephone()


def get_telephone():
    print(threading.current_thread().name + " 取出的手机是", local.telephone + "\n")


if __name__ == '__main__':
    local = threading.local()
    for i in range(3):
        thread = threading.Thread(target=set_telephone, name='学生' + str(i), args=('手机' + str(i),))
        thread.start()

运行结果如下：

学生0 放入的手机是手机0

学生1 放入的手机是手机1

学生2 放入的手机是手机2

学生1 取出的手机是手机1

学生0 取出的手机是手机0

学生2 取出的手机是手机2

可见，每个学生放入的手机与最终取出的手机一致。那么threading的local()函数如何实现这一效果？推测其内部建立了手机与主人之间的映射关系，通过主人的唯一标识来检索对应的手机。

3. 模拟实现local()的功能，创建一个箱子

基于上述推测，我们需定义一个全局字典来存储每个学生放入的手机，字典键为线程ID，值为指定的键值对。示例代码如下：

import threading
import time

global_goods_dict = {}

# {
#     "线程ID":{"telephone":"放入的具体手机"},
#     "线程ID":{"telephone":"放入的具体手机"},
#     "线程ID":{"telephone":"放入的具体手机"}
#
# }

def set_telephone(telephone):
    # 获取线程ID
    thread_id = threading.get_ident()
    global_goods_dict[thread_id] = {}
    global_goods_dict[thread_id]["telephone"] = telephone
    print(threading.current_thread().name + " 放入的手机是", telephone)
    time.sleep(1)
    get_telephone()


def get_telephone():
    thread_id = threading.get_ident()
    print(threading.current_thread().name + " 取出的手机是", global_goods_dict[thread_id]["telephone"])


if __name__ == '__main__':
    for i in range(3):
        thread = threading.Thread(target=set_telephone, name='学生' + str(i), args=('手机' + str(i),))
        thread.start()

运行结果与前述一致。此处定义全局字典global_goods_dict，其键为线程ID，确保每个线程仅能获取自己设置的数据。字典的值同样为字典，因为一个线程可能存储多个值。其中global_goods_dict[thread_id]["telephone"] = telephone的作用等价于上例中的local.telephone = telephone。尽管此方法有效，但使用略显繁琐，能否像local()函数那样简洁丝滑？

4. 简化代码操作，进一步模拟实现local()函数

我们可以将全局global_goods_dict字典封装在一个类中，使该类自动处理值的设置。

class MyBox:
    box = {}

    def __setattr__(self, key, value):
        thread_id = threading.get_ident()
        # 单元格已存在
        if thread_id in MyBox.box:
            MyBox.box[thread_id][key] = value
        else:
            MyBox.box[thread_id] = {key: value}

    def __getattr__(self, item):
        thread_id = threading.get_ident()
        return MyBox.box[thread_id][item]


def set_telephone(telephone):
    myBox.telephone = telephone
    print(threading.current_thread().name + " 放入的手机是", myBox.telephone + "\n")
    time.sleep(1)
    get_telephone()


def get_telephone():
    print(threading.current_thread().name + " 取出的手机是", myBox.telephone + "\n")


if __name__ == '__main__':
    myBox = MyBox()
    for i in range(3):
        thread = threading.Thread(target=set_telephone, name='学生' + str(i), args=('手机' + str(i),))
        thread.start()

运行结果同上。此处通过MyBox类封装了名为box的字典，其键为当前线程ID，值为变量名及对应值组成的键值对。当执行set_telephone方法的myBox.telephone = telephone时，实际调用MyBox的__setattr__方法，参数key为telephone，value为“手机xx”。当调用myBox.telephone时，实际调用__getattr__方法，传入参数item为telephone。取值时首先获取当前线程ID。