FastAPI+Gunicorn如何实现多进程间共享变量 以及 depends中的类变量导致跨请求会话传递数据的问题_gunicorn fastapi

一、FastAPI 框架+Gunicorn如 何实现多进程间共享变量

        使用 FastAPI 框架开发的项目结合 Gunicorn 部署，采用 tiangolo/uvicorn-gunicorn-fastapi 部署项目启动后会看到项目启动了多个进程，多数情况下我们只需要考虑每个进程里的业务运行即可，但有时也会碰到情况一次启动的多个进程我们只允许一个进程执行一个特殊任务，从而避免任务重复进行，有什么好办法呢？当然引入第三方服务比如redis、mysql这些来做锁处理也是一个方法，但不想专门为这一个项去增加额外的redis服务，也不想给mysql增加这个工作负担，想能不能有进程间共享来解决。也看到Flask gunicorn中使用–preload实现工作进程间的内存共享，都未解决我的问题。

    当然，我这里还有一个前提，就是这里并不是集群模式，如果是多服务器，则有必要使用redis了。这里是在一台服务器中执行。

    在此条件下，我尝试了多种方法，一开始在fastapi框架中使用一些全局变量、Mmap、multiprocessing等都不能，也看过一些在gunicorn中使用--reload之类的方法，都未能有好的实现。后来使用了文件锁 fcntl，这个方法挺适合。结合 gunctorn 中的 prestart.sh 。

第1步：在 prestart.sh 中初始化

    首先我们通过：gunctorn 中的 prestart.sh 在每次项目启动时执行一个命令，在 conf/flag.lock 文件中写入字符串0，以保证每次启动都能执行任务。

echo '0' > conf/flag.lock

第2步：在框架中变更文件

这一步中我们需要在 FastAPI 框架中 @app.on_event("startup") 的startup方法里添加处理。

import fcntl, os, random, time
bgfile = 'conf/flag.lock'
 
with open(bgfile, 'r+') as fn:
 fcntl.flock(fn.fileno(), fcntl.LOCK_EX)
 data = fn.read()
 data = data.strip()
 
  if str(data) == '0': 
 fn.seek(0)
 fn.truncate()
 fn.write('1')
 # 单进程执行任务开始
 # 任务...
 else:
 pass

3. 关于Windows和Linux的区别

        在使用中我又发现 fcntl 不能在windows上运行，于是又找了找 fcntl  的替代方法。使用基于 Portalocker  的文件锁能很好的满足我的需求。git 地址：

https://github.com/WoLpH/portalocker Portalocker  - Cross-platform locking library。

    portalocker是一个跨平台的文件锁库，使用起来也非常方便，修改如下：

import portalocker, os, random, time
with open(bgfile, 'r+') as fn:
 portalocker.lock(fn, portalocker.LockFlags.EXCLUSIVE)
 
 data = fn.read()
 data = data.strip()
 
  if str(data) == '0': 
 fn.seek(0)
 fn.truncate()
 fn.write('1')
 fn.flush()
        os.fsync(fn.fileno())
 
 # 单进程执行任务开始
 # 任务...
 else:
 pass

二、FastAPI中的depends中的类变量导致跨请求会话传递数据的问题 

    使用FastAPI搭建了一个python服务端项目里，在route的请求处理方法中使用depends来判断用户的登录状态。遇到了depends中的类变量导致跨请求会话传递数据的问题。

    依赖项是FastAPI 提供的简单易用、功能强大的依赖注入系统。可以让开发人员轻松地把组件集成至 FastAPI。依赖注入非常适合共享业务逻辑、共享数据库连接、实现安全、验证、角色权限等场景，这些场景能复用相同的代码逻辑，而使用依赖注入，将代码重复最小化。依赖项可以是一个函数，也可以是一个类，官方的一个使用类做depends的示例如下：

from fastapi import Depends, FastAPI
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
 
class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit
 
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

    之前的项目中我使用过函数作为depends，有些不足，这次我用的类。在上面的例子中FastAPI 调用 CommonQueryParams 类。这将创建该类的一个 "实例"，该实例将作为参数 commons 被传递给你的函数。我这里是在route的请求处理方法中使用depends来判断用户的登录状态，如部分代码示意如下：

# 接口函数
@router.post("/userinfo", summary="用户信息接口")
async def userinfo(
	auto_login: AutoLogin = Depends(AutoLogin)
)
 
# AutoLogin类代码示意
class AutoLogin:
	user = {}
    def __init__(self, x_token: Optional[str] = Header(None), request: Request = None):
	
		# 判断header中的x_token的有效性，
		# 如有效则提取并保存user字典存至AutoLogin的user中。
		AutoLogin.user = user

    AutoLogin是一个depends类，其中会判断header中的x_token的有效性，如有效则提取并保存user字典和相关token字典至AutoLogin的类变量user和token中。在应用中发现用户请求未带有x_token的情况下API返回了用户的token数据，调试排查发现存在这种情况，上次用户有一个请求中带了x_token，然而在其接下来的一次请求中没有带x_token的前提下API接口获取到了上次提交的x_token，即导致了跨请求跨会话的数据传递。因为上面的AutoLogin类中使用了类变量导致的，类变量在整个实例化的对象中是公用的。类变量定义在类中且在函数体之外，有些类似静态属性的作用，但各种语言的执行逻辑不一样会出现不同的结果，在有些语言中虽然是静态属性但只会在某次请求中共享，而python中这种类变量则能跨请求传递共享，导致问题。

    因此碰到这样的问题，不要在AutoLogin类中定义类变量，而要在类方法中使用self.类名来创建实例变量。