源码: tests/local.py

# -.- coding:utf-8 -.-
import unittest
import threading


class TestLocal(unittest.TestCase):

    def test_create_local_instance(self):
        data = threading.local()
        self.assertEqual(data.__dict__, {})

    def test_add_kv_to_local_instance(self):
        data = threading.local()
        data.language = "python"
        data.version = "2.7.13"
        self.assertEqual(data.__dict__,
                         {"language": "python", "version": "2.7.13"})

    def test_del_kv_to_local_instance(self):
        data = threading.local()
        data.language = "python"
        data.version = "2.7.13"
        del data.language
        self.assertEqual(data.__dict__, {"version": "2.7.13"})

    def test_cross_threads_access_local(self):
        data = threading.local()
        data.language = "python"

        def other_thread():
            # 在其他线程中, local 对象是一个空对象,
            # 在该线程中创建的属性值, 其他线程(包括主线程)
            # 都无法访问到, 只有当前线程能访问到.
            #
            # local 就好像是一个大字典, 每个线程一个命名空间,
            # 表现形式: {"MainThread": {"language": "python"},
            #            "Thread-1": {"other_thread": 1}}
            #
            # 备注:
            # local 只需在全局作用域中创建一次, 其他线程即可引用和访问.
            # 无需在每个线程作用域中单独创建一次实例.
            self.assertEqual(data.__dict__, {})

            current_thread_id = threading.current_thread().ident
            data.other_thread = current_thread_id
            self.assertEqual(data.__dict__,
                             {"other_thread": current_thread_id})

        t = threading.Thread(target=other_thread, args=())
        t.start()
        t.join()
        self.assertEqual(data.__dict__, {"language": "python"})

测试: tests/main.py

import unittest


TEST_MODULE = [
    "ln_threading.tests.local",
]


if __name__ == '__main__':
    suite = unittest.defaultTestLoader.loadTestsFromNames(TEST_MODULE)
    runner = unittest.TextTestRunner(verbosity=2)
    runner.run(suite)