Python 实现 ThreadLocal

发布时间 2023-07-10 15:40:09作者: katsute_不语
import threading
from threading import get_ident
import time

class Context:

    def __init__(self):
        object.__setattr__(self, '__global_context__', dict())
        object.__setattr__(self, '__get_ident__', get_ident)

    def __setattr__(self, key, value):
        try:
            self.__global_context__[self.__get_ident__()][key] = value
        except KeyError:
            self.__global_context__[self.__get_ident__()] = {key: value}

    def __getattr__(self, item):
        try:
            return self.__global_context__[self.__get_ident__()][item]
        except KeyError:
            return None

    def __delattr__(self, item):
        try:
            self.__global_context__[self.__get_ident__()].pop(item, None)
        except KeyError:
            return None

    def __repr__(self):
        return str(self.__global_context__.get(self.__get_ident__()))


ctx = Context()
ctx.a = 1
ctx.b = 2
ctx.c = 2


def as_fun(i):
    ctx.a = i
    print(ctx)
    time.sleep(1)


print(ctx.__dict__)
threads = []
for _ in range(5):
    t = threading.Thread(target=as_fun, args=(_,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()

print(ctx.__dict__)