🍖Flask自定义 local 对象 (实现并发处理请求)

自定义 local 对象 (实现并发处理请求)

1.思考及需求

• 要实现并发效果, 每一个请求进来的时候我们都开启一个进程, 这显然是不合理的, 于是就可以使用线程
• 如果我们的需求是每个线程都对变量 num 进行设值, 并打印其线程号, 其效果如下 :

from threading import Thread, get_ident  # 可以获取线程id
import time

num = -1

def task(arg):
    global num
    num = arg
    time.sleep(2)
    print(num, f'线程{get_ident()}')


for i in range(10):
    t = Thread(target=task, args=(i,))
    t.start()
"""
9 线程12244
9 线程13324
9 线程7972
9 线程134249 
9 线程12476
9 线程13408
...
"""

很明显, 数据改乱了, 是个线程更改的都是同一份数据, 对数据产生了不安全性

为了让他们的更改相互隔离, 于是就可以 :

1. 加锁 : 不使用该方法, 加锁的思路是多个线程要真正实现共用一个数据, 而我们是要做请求对象的并发, 实现的是该线程对于请求对象这部分内容有任何修改并不影响其他线程
2. 所以使用 local 对象将要修改的数据复制一份，使得每个数据互不影响

2.使用 local 对象

• local对象只支持线程，不支持协程，生成自己的local对象，各改各的，互不影响

from threading import Thread
from threading import local
from threading import get_ident
import time

# local对象,当识别到新的进程会为其开辟一块新的内存空间, 相当于每个线程都对该值进行了拷贝
local_obj = local()
'''
{'线程1':{'value':1},'线程2':{'value':1},'线程3':{'value':1},'线程4':{'value':1}}
'''

def task(arg):
    local_obj.value = arg
    time.sleep(2)
    print(local_obj.value,f'线程号:{get_ident()}')

for i in range(10):
    t = Thread(target=task, args=(i,))
    t.start()
    
"""
2 线程号:11896
3 线程号:8796
9 线程号:3100
8 线程号:9188
....
"""

3.通过字典自定义 local 对象

• 面向过程式编写

from threading import get_ident, Thread

storage = {}  # 初始化一个字典
'''
{'线程1':{'value':1},'线程2':{'value':1},'线程3':{'value':1},'线程4':{'value':1}}
'''

def set(k, v):
    # 获取线程id
    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()

"""
0
1
2
3
4
5
6
7
8
9
"""

• 面向对象式编写

from threading import get_ident, Thread

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

from threading import get_ident, Thread

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()

上面已经实现了 local 的功能, 但存在一个问题 : 如果我们想生成多个local对象，但是会导致多个local对象所管理的线程设置的内容都放到了类属性 storage = { }里面, 于是我们可以想到将 storage 设置成对象属性

• 将storage = { }设置成对象属性, 实现每一个local对象所对应的线程设置的内容都放到自己的storage里面

from threading import get_ident, Thread

class Local(object):
    def __init__(self):
        object.__setattr__(self, 'storage', {})
        # self.__setattr__('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
    print(obj.val)

for i in range(10):
    t = Thread(target=task, args=(i,))
    t.start()

4.使用协程 | 线程来实现请求并发

• 协程属于应用级别的, 协程会替代操作系统自动切换遇到 IO的任务或者运行级别低的任务, 而应用级别的切换速度远高于操作系统的切换
• 在flask中为了实现这种并发需求, 依赖于werkzeug包, 我们导入werkzeug下的local查看其源码

try:
    from greenlet import getcurrent as _get_ident  # 获取协程唯一标识
except ImportError:
    from threading import get_ident as _get_ident  # 获取线程唯一标识

• 发现最开始导入线程和协程的唯一标识的时候统一命名为_get_ident，并且先导入协程模块的时候如果报错说明不支持协程，就会去导入线程的_get_ident，这样无论是只有线程运行还是协程运行都可以获取唯一标识
• 我们使用该方法实现协程 | 线程并发处理请求

try:
    from greenlet import getcurrent as get_ident  # 获取协程唯一标识
except Exception as e:
    from threading import get_ident  # 获取进程唯一标识

from threading import Thread

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()