应用调度¶

    应用调度与 模块化 的最大不同在于应用调度中的每个应用是完全独立的,它们以各自的配置运行,并在 WSGI 层面被调度。

    下面所有的技术说明和举例都归结于一个可以运行于任何 WSGI 服务器的 对象。对于生产环境,参见 。对于开发环境,Werkzeug 提供了一个内建开发服务器,它使用werkzeug.serving.run_simple() 来运行:

    注意 不能用于生产环境,生产环境服务器参见 成熟的 WSGI 服务器

    1. from flask import Flask
    2. from werkzeug.serving import run_simple
    3.  
    4. app = Flask(__name__)
    5. app.debug = True
    6.  
    7. @app.route('/')
    8. def hello_world():
    9. return 'Hello World!'
    10.  
    11. if __name__ == '__main__':
    12. run_simple('localhost', 5000, app,
    13. use_reloader=True, use_debugger=True, use_evalex=True)

    如果你想在同一个 Python 解释器中运行多个独立的应用,那么你可以使用 。其原理是:每个独立的 Flask应用都是一个合法的 WSGI 应用,它们通过调度中间件组合为一个基于前缀调度的大应用。

    假设你的主应用运行于 / ,后台接口位于 :

    有时候你可能需要使用不同的配置来运行同一个应用的多个实例。可以把应用创建过程放在一个函数中,这样调用这个函数就可以创建一个应用的实例,具体实现参见应用工厂 方案。

    WSGI 层是完美的抽象层,因此可以写一个你自己的 WSGI 应用来监视请求,并把请求分配给你的 Flask 应用。如果被分配的应用还没有创建,那么就会动态创建应用并被登记下来:

    1. from threading import Lock
    2.  
    3. class SubdomainDispatcher(object):
    4.  
    5. def __init__(self, domain, create_app):
    6. self.create_app = create_app
    7. self.lock = Lock()
    8. self.instances = {}
    9.  
    10. def get_application(self, host):
    11. host = host.split(':')[0]
    12. assert host.endswith(self.domain), 'Configuration error'
    13. subdomain = host[:-len(self.domain)].rstrip('.')
    14. with self.lock:
    15. app = self.instances.get(subdomain)
    16. if app is None:
    17. app = self.create_app(subdomain)
    18. self.instances[subdomain] = app
    19. return app
    20.  
    21. def __call__(self, environ, start_response):
    22. app = self.get_application(environ['HTTP_HOST'])
    23. return app(environ, start_response)

    调度器示例:

    根据 URL 的路径调度非常简单。上面,我们通过查找 Host 头来判断子域,现在只要查找请求路径的第一个斜杠之前的路径就可以了:

    1. from threading import Lock
    2. from werkzeug.wsgi import pop_path_info, peek_path_info
    3.  
    4. class PathDispatcher(object):
    5. def __init__(self, default_app, create_app):
    6. self.default_app = default_app
    7. self.create_app = create_app
    8. self.lock = Lock()
    9. self.instances = {}
    10.  
    11. def get_application(self, prefix):
    12. with self.lock:
    13. app = self.instances.get(prefix)
    14. if app is None:
    15. app = self.create_app(prefix)
    16. if app is not None:
    17. self.instances[prefix] = app
    18. return app
    19.  
    20. def __call__(self, environ, start_response):
    21. app = self.get_application(peek_path_info(environ))
    22. if app is not None:
    23. pop_path_info(environ)
    24. else:
    25. app = self.default_app
    26. return app(environ, start_response)