gunicorn + Flask架构中使用多进程全局锁

Posted 2022-05-26 上帝De助手

有之前的认识WSGI和WSGI的前世今世之后，现在就可以介绍如何在gunicorn + Flask架构模式下，在Flask处理线程中使用全局锁。

 

说到锁在Python中也有很多锁，最常见用的就是多进程锁（multiprocessing.Lock）和多线程锁（threading.Lock）。正常情况下，我们是可以直接使用这些锁的。多进程锁可以在多个子进程中实现锁定临界资源的功能，而多线程锁则只在子线程中可以锁定临界资源。

 

而一旦你使用了gunicorn + Flask的架构。gunicorn就会启动多个worker子进程，每个子进程可以看做是一个独立的Flask进程。现在需要在所有worker进程中的Flask应用内申请锁资源，并且该锁资源需要在其它worker中是互斥的。

 

在不改变gunicorn源码的情况下，我们无法在主进程中创建一个锁，之后在子进程中直接使用；并且在gunicorn调用flask接口的时候也未提供传入额外参数的接口。所以在一开始的时候，并未找到让各Falsk子进程共享锁的方法。通过网上的相关查询，觅得另外的解决方案：通过实现一个脱离Flask进程的外部锁，比如：db记录锁、redis记录锁、文件锁等方式。最终我选择了使用文件锁来解决Flask进程之间的锁共享问题。文件锁代码如下：

 

#!/usr/bin/env python
# coding=utf-8
#
# File Lock For Multiple Process
import os
import time

WINDOWS = 'windows'
LINUX = 'linux'
SYSTEM = None

try:
    import fcntl
    SYSTEM = LINUX
except:
    SYSTEM = WINDOWS

class Lock(object):
    @staticmethod
    def get_file_lock():
        return FileLock()

class FileLock(object):
    def __init__(self):
        lock_file = 'FLASK_LOCK'
        if SYSTEM == WINDOWS:
            lock_dir = os.environ['tmp']
        else:
            lock_dir = '/tmp'

        self.file = '%s%s%s' % (lock_dir, os.sep,lock_file)
        self._fn = None
        self.release()

    def acquire(self):
        if SYSTEM == WINDOWS:
            while os.path.exists(self.file):
                time.sleep(0.01)    #wait 10ms
                continue

            with open(self.file, 'w') as f:
                 f.write('1')
        else:
            self._fn = open(self.file, 'w')
            fcntl.flock(self._fn.fileno(), fcntl.LOCK_EX)
            self._fn.write('1')

    def release(self):
        if SYSTEM == WINDOWS:
            if os.path.exists(self.file):
                os.remove(self.file)
        else:
            if self._fn:
                try:
                    self._fn.close()
                except:
                    pass

该文件锁类可以提供一个基于外部文件资源的锁，在Windows环境下其实并不能完全锁定资源，小概率的情况下会锁定失败。但在linux下则会正常工作，因为在linux下使用了文件锁模块，它可以确保加锁过程是原子操作。这个锁的使用方法也很简单：

 

 

from flask import Flask, current_app
from lock import Lock

app = Flask(__name__)
app.lock = Lock.get_file_lock()     ##给app注入一个外部锁

@app.route("/")
def hello():
	current_app.lock.acquire()  ##获取锁
	current_app.lock.release()  ##释放锁
    return "Hello World!"

只要在flaskapp中注入文件锁，此外在其它模块中通过current_app模块即可获取到该锁。

 

 

通常上述方法就可以解决Flask子进程之间的锁共享问题了，但是如果你的环境必须是windows的，并且不能有任何的锁定失败情况，那么你还是得查找其它可用的方法。黄天不负有心人，gunicorn虽然默认没有说支持注入锁到flask进程的接口，但是它还是需要与flask通信的。基于前面WSGI的文章，我们可以知道它们通信的方式其实就是调用WSGI的接口。而该接口支持2个参数：

 

1. 第一个参数：当前请求的相关信息，比如：头信息、请求参数
2. 第二个参数：一个返回状态码和响应头的回调函数

 

仔细想想第一个参数可能还是可以利用的，所以如果我们在这之前把锁对象也能塞到这个参数中，那我们其实就可以获取到外部的锁了。因为该参数收集的基本上是请求头信息，所以如果我们可以把锁把塞到请求头，会如何呢？ 

 

正好gunicorn有server hook回调函数，可以支持我们在server和worker工作的期间进行相关的对象操作。其中一个就是操作请求对象（request），所以我们现在就可以往请求头中添加Lock对象了。添加之后能不能传递到flask子进程呢？测试一下即可。

首先，创建一个WSGI的app应用文件app.py，内容如下：

 

#app.py
def app(environ, start_response):
	print environ
	data = b"Hello, World!\\n"
	start_response("200 OK", [
		("Content-Type", "text/plain"),
		("Content-Length", str(len(data)))
	])
	return iter([data])

 

接着，创建一个gunicorn的配置文件cnf.py，内容如下：

 

#cnf.py
import multiprocessing

lock = multiprocessing.Lock()

def pre_request(worker, req):
    req.headers['FLASK_LOCK'] = lock

pre_request = pre_request
bind = '0.0.0.0:8000'
workers = multiprocessing.cpu_count()
worker_class = 'gevent'

之后，我们就可以启动gunicorn来测试下lock对象是否被传递给了WSGI接口参数中。执行如下命令：

 

gunicorn -c cnf.py fapp:app

最后，还需要在本地浏览器中访问http://localhost:8000/来查看执行结果。很开心的是这个lock对象【HTTP_FLASK_LOCK】被当作正常的请求头信息传递给了app接口。打印的效果如下：

 

'HTTP_COOKIE': 'token=b3bdd0b4a64d7bd14851067a2775a71955d3ba8feyJpZCI6IDJ9; session=eyJ0b2tlbiI6eyJpZCI6Mn19.DO1Geg.LrqGWGQXeBEwKy-zw49rbi5Q4XY', 'SERVER_SOFTWARE': 'gunicorn/19.7.1', 'SCRIPT_NAME': '', 'REQUEST_METHOD': 'GET', 'PATH_INFO': '/', 'SERVER_PROTOCOL': 'HTTP/1.1', 'QUERY_STRING': '', 'HTTP_FLASK_LOCK': <Lock(owner=None)>, 'HTTP_USER_AGENT': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (Khtml, like Gecko) Chrome/62.0.3202.94 Safari/537.36', 'HTTP_CONNECTION': 'keep-alive', 'REMOTE_PORT': '10648', 'SERVER_NAME': '0.0.0.0', 'REMOTE_ADDR': '172.16.1.167', 'wsgi.url_scheme': 'http', 'SERVER_PORT': '88', 'wsgi.input': <gunicorn.http.body.Body object at 0x131c650>, 'HTTP_HOST': '172.16.1.156:88', 'wsgi.multithread': True, 'HTTP_UPGRADE_INSECURE_REQUESTS': '1', 'HTTP_CACHE_CONTROL': 'max-age=0', 'HTTP_ACCEPT': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8', 'wsgi.version': (1, 0), 'RAW_URI': '/', 'wsgi.run_once': False, 'wsgi.errors': <gunicorn.http.wsgi.WSGIErrorsWrapper object at 0x131c7d0>, 'wsgi.multiprocess': True, 'HTTP_ACCEPT_LANGUAGE': 'zh-CN,zh;q=0.9', 'gunicorn.socket': <socket at 0x12be650 fileno=15 sock=172.16.1.156:88 peer=172.16.1.167:10648>, 'wsgi.file_wrapper': <class 'gunicorn.http.wsgi.FileWrapper'>, 'HTTP_ACCEPT_ENCODING': 'gzip, deflate'

 

所以，结论是如果我们在gunicorn的server hook回调函数中对request对象进行追加内容，那么gunicorn会原封不动的给传递给WSGI接口函数。

 

 

那么，还有一个问题是在Flask中，要如何获取这个lock对象呢？因为Flask已经对WSGI接口进行了封装，我们正常是无法访问其WSGI接口函数的参数。而由于gunicorn传递的参数都是请求头信息，所以第一时间可想到的可能对象，应该就是Flask的request对象。因为request对象中有headers对象，它就是存放当前请求的头信息，与之前添加lock时追加到headers对象相呼应。所以可以来测试一把。
新建一个Flask应用文件fapp.py，内容如下：

 

##fapp.py
from flask import Flask, request
app = Flask(__name__)

@app.route("/")
def hello():
    print request.headers
    return "Hello World!"

启动gunicorn命令：

gunicorn -c cnf.py fapp:app

本地浏览器访问http://localhost:8000查看执行结果：又一次很开心的开到了lock【Flask-Lock】的存在。打印结果如下：

 

 

Flask-Lock: <Lock(owner=None)>
User-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36
Connection: keep-alive
Host: 172.16.1.156:88
Upgrade-Insecure-Requests: 1
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8
Accept-Language: zh-CN,zh;q=0.9
Accept-Encoding: gzip, deflate

但是，在进一步读取Flask-Lock头信息时，发现其值却是一个str类型。所以，我们并没有真正的获取到lock对象。黄天不负有心人，在我再一次查看request对象的成员时，很鸡贼的发现有一个environ成员。于是毫不犹豫的打印出来，发现这个environ其实就是gunicorn调用WSGI接口函数时传递的那个environ参数。
所以在Flask中正确获取全局锁的姿势是：

lock = request.environ['HTTP_FLASK_LOCK']

完整的代码内容如下：

##fapp.py
from flask import Flask, request
app = Flask(__name__)

@app.route("/")
def hello():
    lock = request.environ['HTTP_FLASK_LOCK']
    lock.acquire()
    ##do something
    lock.release()
    return "Hello World!"

再次重启gunicorn命令， 你的业务代码就可以正常获取全局的进程锁了。关于学习Python的更多文章，请扫描下方二维码。