13518219792
class Foo(object):
def __init__(self):
# self.storage = {}
object.__setattr__(self, 'storage', {})
def __setattr__(self, key, value):
self.storage[key] = value
def __getattr__(self, item):
return self.storage.get(item)
cls = Foo() #創建對象
cls.x1 = 123 #調用setattr方法
cls.x1 #調用getattr方法
import threading
from threading import get_ident
def task():
ident = get_ident() # 線程的唯一標識 線程id
print(ident)
for i in range(10):
t = threading.Thread(target=task)
t.start()
每一個request都是一個線程，每個線程有自己唯一的線程ID，利用這個線程ID作為標識，來存儲request對象。
這樣的話就可以為每個線程開辟獨立的空間互不影響。
創新互聯專注于邕寧網站建設服務及定制，我們擁有豐富的企業做網站經驗。 熱誠為您提供邕寧營銷型網站建設，邕寧網站制作、邕寧網頁設計、邕寧網站官網定制、重慶小程序開發公司服務，打造邕寧網絡公司原創品牌,更為您提供邕寧網站排名全網營銷落地服務。
import threading
from threading import get_ident
'''
storage = {
1111:{x1:1}
1112:{x1:2}
1113:{x1:3}
1114:{x1:4}
}
'''
class Local(object):
def __init__(self):
object.__setattr__(self, 'storage', {}) # 相當于self.storage = {}
def __setattr__(self, key, value):
ident = get_ident()
if ident in self.storage: # 如果ident在字典里面就直接修改就行了
self.storage[ident][key] = value
else: # 不在字典里面就添加
self.storage[ident] = {key:value}
def __getattr__(self,item):
ident = get_ident()
if ident not in self.storage:
return
return self.storage[ident].get(item)
local = Local()
def task(num):
local.x1 = num
print(local.x1)
if __name__ == '__main__':
for i in range(5):
t = threading.Thread(target=task, args=(i,))
t.start()
import threading
from threading import get_ident
'''
storage = {
1111:{x1:[1]}
1112:{x1:[]}
1113:{x1:[]}
1114:{x1:[]}
}
維護成一個棧
'''
class Local(object):
def __init__(self):
object.__setattr__(self, 'storage', {}) # 相當于self.storage = {}
def __setattr__(self, key, value):
ident = get_ident()
if ident in self.storage: # 如果ident在字典里面就直接修改就行了
self.storage[ident][key].append(value)
else: # 不在字典里面就添加
self.storage[ident] = {key:[value]}
def __getattr__(self,item):
ident = get_ident()
if ident not in self.storage:
return
return self.storage[ident][item][-1]
local = Local()
def task(num):
local.x1 = num
print(local.x1)
if __name__ == '__main__':
for i in range(5):
t = threading.Thread(target=task, args=(i,))
t.start()
flask的locak對象
class Local:
__slots__ = ("_storage",)
def __init__(self) -> None:
object.__setattr__(self, "_storage", ContextVar("local_storage"))
@property
def __storage__(self) -> t.Dict[str, t.Any]:
warnings.warn(
"'__storage__' is deprecated and will be removed in Werkzeug 2.1.",
DeprecationWarning,
stacklevel=2,
)
return self._storage.get({}) # type: ignore
@property
def __ident_func__(self) -> t.Callable[[], int]:
warnings.warn(
"'__ident_func__' is deprecated and will be removed in"
" Werkzeug 2.1. It should not be used in Python 3.7+.",
DeprecationWarning,
stacklevel=2,
)
return _get_ident # type: ignore
@__ident_func__.setter
def __ident_func__(self, func: t.Callable[[], int]) -> None:
warnings.warn(
"'__ident_func__' is deprecated and will be removed in"
" Werkzeug 2.1. Setting it no longer has any effect.",
DeprecationWarning,
stacklevel=2,
)
def __release_local__(self) -> None:
__release_local__(self._storage)
def __getattr__(self, name: str) -> t.Any:
values = self._storage.get({})
try:
return values[name]
except KeyError:
raise AttributeError(name) from None
def __setattr__(self, name: str, value: t.Any) -> None:
values = self._storage.get({}).copy()
values[name] = value
self._storage.set(values)
def __delattr__(self, name: str) -> None:
values = self._storage.get({}).copy()
try:
del values[name]
self._storage.set(values)
except KeyError:
raise AttributeError(name) from None
flask的localstack對象
class LocalStack:
def __init__(self) -> None:
self._local = Local()
def __release_local__(self) -> None:
self._local.__release_local__()
@property
def __ident_func__(self) -> t.Callable[[], int]:
return self._local.__ident_func__
@__ident_func__.setter
def __ident_func__(self, value: t.Callable[[], int]) -> None:
object.__setattr__(self._local, "__ident_func__", value)
def __call__(self) -> "LocalProxy":
def _lookup() -> t.Any:
rv = self.top
if rv is None:
raise RuntimeError("object unbound")
return rv
return LocalProxy(_lookup)
def push(self, obj: t.Any) -> t.List[t.Any]:
"""Pushes a new item to the stack"""
rv = getattr(self._local, "stack", []).copy()
rv.append(obj)
self._local.stack = rv
return rv # type: ignore
def pop(self) -> t.Any:
"""Removes the topmost item from the stack, will return the
old value or `None` if the stack was already empty.
"""
stack = getattr(self._local, "stack", None)
if stack is None:
return None
elif len(stack) == 1:
release_local(self._local)
return stack[-1]
else:
return stack.pop()
@property
def top(self) -> t.Any:
"""The topmost item on the stack. If the stack is empty,
`None` is returned.
"""
try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
在flask中有個local類，他和threading.local的功能一樣，為每個線程開辟空間來存儲數據，他們兩個的內部實現機制一樣，內部維護一個字典，以線程（協程）ID為key，進行數據隔離如：
__storage__ = {
1111:{'k1':123}
}
obj = local()
obj.k1 = 123
在flask中還有localstack的類，他內部會依賴local對象，local對象負責存儲數據，localstack對象用于將local中的值維護成一個棧。
__storage__ = {
1112:{'stack':[k1,]}
}
obj = LocalStack()
obj.push('k1')
obj.pop()
obj.top
context locals
__storage__ = {
'1111':{'stack':['RequestContext(request, session)',]},
'1112':{'stack':['RequestContext(request, session)',]}
}
__storage__ = {
'1111':{'stack':['AppContext(app, g)',]},
'1112':{'stack':['AppContext(app, g)',]}
}
_request_ctx_stack = LocalStack() # 請求上下文
_app_ctx_stack = LocalStack() # 應用上下文
