Flask之线程与协程

Posted 2020-12-16 shenjianping

一、前言

还记得在flask中是怎么引入request对象的吗？没错是通过：

from flask import request

　　那么，这样全局引入的，势必会存在下面的问题，如果多个用户同时发送请求，一个request对象会被多个用户修改，最后大家拿到的返回值就都是最后一个用户的返回值，那么应该如何解决这种问题呢？

（一）threadinglocal

import threading

local_values = threading.local()


def func(num):
    local_values.num_ = num
    print(local_values.num_, threading.current_thread().name)


for i in range(10):
    t = threading.Thread(target=func, args=(i,), name=‘线程%s‘ % i)
    t.start()
"""
输出：
0 线程0
1 线程1
2 线程2
3 线程3
4 线程4
5 线程5
6 线程6
7 线程7
8 线程8
9 线程9
"""

　　通过上面的例子可以看出threading.local()对象的作用就是为每一个线程开辟一个独有的空间，用于保存每一个线程自己独有的值，这样不至于一个值被多个线程改来改去造成数据的混乱。

（二）request对象

对于flask中的request有下面三种情况：

• 单进程、单线程，只需要基于全局变量实现即可
• 单进程、多线程，此时需要threading.local()对象来实现，为每一个请求的request单独开辟一个空间，这样不会造成request对象的混乱
• 单进程、单线程、多协程，这种情况threading.local()对象是无法实现的，因为线程中协程的资源都是共享的

那么，如果在flask中支持协程，应该怎么实现呢？

二、自定义支持协程和协程的Local对象

如果要支持线程，可以自定义类似threading.local()对象，那么它是在threading.local()对象的基础上进一步强化，可以支持协程。

import threading
from threading import get_ident

class Local:

    def __init__(self):
        self.storage = {}    #会生成字典 {5936: {‘num_‘: 3}, 7796: {‘num_‘: 6}, 7056: {‘num_‘: 7}...}
        self.get_ident = get_ident


    def set(self,k,v):
        ident = self.get_ident()
        origin = self.storage.get(ident)
        if not origin:
            origin = {k:v}
        else:
            origin[k] = v
        self.storage[ident] = origin

    def get(self,k):
        ident = self.get_ident()
        origin = self.storage.get(ident)
        if not origin:
            return None
        return origin.get(k,None)


local_values = Local()


def task(num):
    local_values.set(‘num_‘,num)
    print(local_values.get(‘num_‘),threading.current_thread().name)

if __name__ == ‘__main__‘:
    for i in range(10):
        t = threading.Thread(target=task,args=(i,),name=‘线程%s‘%i)
        t.start()

"""
输出
0 线程0
1 线程1
2 线程2
3 线程3
4 线程4
5 线程5
6 线程6
7 线程7
8 线程8
9 线程9

"""

在flask的源码werkzeug.local.py文件中：

# since each thread has its own greenlet we can just use those as identifiers
# for the context.  If greenlets are not available we fall back to the
# current thread ident depending on where it is.
try:
    from greenlet import getcurrent as get_ident
except ImportError:
    try:
        from thread import get_ident
    except ImportError:
        from _thread import get_ident

class Local(object):
    __slots__ = ("__storage__", "__ident_func__")

    def __init__(self):
        object.__setattr__(self, "__storage__", {})
        object.__setattr__(self, "__ident_func__", get_ident)

    def __iter__(self):
        return iter(self.__storage__.items())

    def __call__(self, proxy):
        """Create a proxy for a name."""
        return LocalProxy(self, proxy)

    def __release_local__(self):
        self.__storage__.pop(self.__ident_func__(), None)

    def __getattr__(self, name):
        try:
            return self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)

    def __setattr__(self, name, value):
        ident = self.__ident_func__()
        storage = self.__storage__
        try:
            storage[ident][name] = value
        except KeyError:
            storage[ident] = {name: value}

    def __delattr__(self, name):
        try:
            del self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)

源码Local对象

参考源码可以自己写的Local对象进行优化：

import threading

try:
    from greenlet import getcurrent as get_ident  # 协程
except ImportError:
    try:
        from thread import get_ident
    except ImportError:
        from _thread import get_ident  # 线程


class Local(object):

    def __init__(self):
        object.__setattr__(self, ‘__storage__‘, {})
        object.__setattr__(self, ‘__ident_func__‘, get_ident)

    def __getattr__(self, name):
        try:
            return self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)

    def __setattr__(self, name, value):
        ident = self.__ident_func__()
        storage = self.__storage__
        try:
            storage[ident][name] = value
        except KeyError:
            storage[ident] = {name: value}

    def __delattr__(self, name):
        try:
            del self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)


local_values = Local()


def task(num):
    local_values.num_ = num  # 通过对象.key=value的方式触发__setattr__方法
    print(local_values.num_, threading.current_thread().name)  # 通过对象.key触发__getattr__方法


if __name__ == ‘__main__‘:
    for i in range(10):
        t = threading.Thread(target=task, args=(i,), name=‘线程%s‘ % i)
        t.start()