補(bǔ)充
pipreqs:
- 項(xiàng)目依賴 pip3 install pipreqs
- 生成依賴文件:pipreqs ./
- 安裝依賴文件:pip3 install -r requirements.txt
函數(shù)和方法
from types import MethodType,FunctionType
class Foo(object):
def fetch(self):
pass
print(isinstance(Foo.fetch,MethodType))
print(isinstance(Foo.fetch,FunctionType)) # True
obj = Foo()
print(isinstance(obj.fetch,MethodType)) # True
print(isinstance(obj.fetch,FunctionType))
threading.local
多個(gè)線程修改同一個(gè)數(shù)據(jù),復(fù)制多份變量給每個(gè)線程用,為每個(gè)線程開辟一塊空間進(jìn)行數(shù)據(jù)存儲
不用threading.local
# 不用local
from threading import Thread
import time
lqz = -1
def task(arg):
global lqz
lqz = arg
# time.sleep(2)
print(lqz)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
threading.local使用
from threading import Thread
from threading import local
import time
from threading import get_ident
# 特殊的對象
lqz = local()
def task(arg):
# 對象.val = 1/2/3/4/5
lqz.value = arg
time.sleep(2)
print(lqz.value)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
通過字典自定義threading.local(函數(shù))
from threading import get_ident,Thread
import time
storage = {}
def set(k,v):
ident = get_ident()
if ident in storage:
storage[ident][k] = v
else:
storage[ident] = {k:v}
def get(k):
ident = get_ident()
return storage[ident][k]
def task(arg):
set('val',arg)
v = get('val')
print(v)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
面向?qū)ο蟀?/h3>
from threading import get_ident,Thread
import time
class Local(object):
storage = {}
def set(self, k, v):
ident = get_ident()
if ident in Local.storage:
Local.storage[ident][k] = v
else:
Local.storage[ident] = {k: v}
def get(self, k):
ident = get_ident()
return Local.storage[ident][k]
obj = Local()
def task(arg):
obj.set('val',arg)
v = obj.get('val')
print(v)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
通過setattr和getattr實(shí)現(xiàn)
from threading import get_ident,Thread
import time
class Local(object):
storage = {}
def __setattr__(self, k, v):
ident = get_ident()
if ident in Local.storage:
Local.storage[ident][k] = v
else:
Local.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return Local.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
每個(gè)對象有自己的存儲空間(字典)
from threading import get_ident,Thread
import time
class Local(object):
def __init__(self):
object.__setattr__(self,'storage',{})
def __setattr__(self, k, v):
ident = get_ident()
if ident in self.storage:
self.storage[ident][k] = v
else:
self.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return self.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
obj.xxx = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
兼容線程和協(xié)程
try:
from greenlet import getcurrent as get_ident
except Exception as e:
from threading import get_ident
from threading import Thread
import time
class Local(object):
def __init__(self):
object.__setattr__(self,'storage',{})
def __setattr__(self, k, v):
ident = get_ident()
if ident in self.storage:
self.storage[ident][k] = v
else:
self.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return self.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
obj.xxx = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
partial偏函數(shù)
#偏函數(shù)的第二個(gè)部分(可變參數(shù)),按原有函數(shù)的參數(shù)順序進(jìn)行補(bǔ)充,參數(shù)將作用在原函數(shù)上,最后偏函數(shù)返回一個(gè)新函數(shù)
from functools import partial
def test(a,b,c,d):
return a+b+c+d
tes=partial(test,1,2)
print(tes(3,4))
session源碼分析
SecureCookieSessionInterface
-open_session
-save_session
from threading import get_ident,Thread
import time
class Local(object):
storage = {}
def set(self, k, v):
ident = get_ident()
if ident in Local.storage:
Local.storage[ident][k] = v
else:
Local.storage[ident] = {k: v}
def get(self, k):
ident = get_ident()
return Local.storage[ident][k]
obj = Local()
def task(arg):
obj.set('val',arg)
v = obj.get('val')
print(v)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
from threading import get_ident,Thread
import time
class Local(object):
storage = {}
def __setattr__(self, k, v):
ident = get_ident()
if ident in Local.storage:
Local.storage[ident][k] = v
else:
Local.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return Local.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
from threading import get_ident,Thread
import time
class Local(object):
def __init__(self):
object.__setattr__(self,'storage',{})
def __setattr__(self, k, v):
ident = get_ident()
if ident in self.storage:
self.storage[ident][k] = v
else:
self.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return self.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
obj.xxx = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
try:
from greenlet import getcurrent as get_ident
except Exception as e:
from threading import get_ident
from threading import Thread
import time
class Local(object):
def __init__(self):
object.__setattr__(self,'storage',{})
def __setattr__(self, k, v):
ident = get_ident()
if ident in self.storage:
self.storage[ident][k] = v
else:
self.storage[ident] = {k: v}
def __getattr__(self, k):
ident = get_ident()
return self.storage[ident][k]
obj = Local()
def task(arg):
obj.val = arg
obj.xxx = arg
print(obj.val)
for i in range(10):
t = Thread(target=task,args=(i,))
t.start()
#偏函數(shù)的第二個(gè)部分(可變參數(shù)),按原有函數(shù)的參數(shù)順序進(jìn)行補(bǔ)充,參數(shù)將作用在原函數(shù)上,最后偏函數(shù)返回一個(gè)新函數(shù)
from functools import partial
def test(a,b,c,d):
return a+b+c+d
tes=partial(test,1,2)
print(tes(3,4))
SecureCookieSessionInterface
-open_session
-save_session