如何为多个WSGI应用配置不同Worker并在单端口运行？

我有两个基于Flask开发的WSGI应用elephant和mouse，希望在单个端口上运行它们，且不同应用使用不同进程：mouse需要多实例，elephant仅需单实例，同时避免繁琐的host:port通信配置。

目前我通过gunicorn分别启动它们，虽能实现Worker配置，但应用会运行在不同端口，命令示例如下：

elephant_cmd = ['gunicorn', '-b', '0.0.0.0:5009', '-w', '10', 'elephant:app', '-n', 'worker', '--access-logfile', '-']
mouse_cmd = ['gunicorn', '-b', '0.0.0.0:5001', '-w', '2', 'mouse:app', '-n', 'model', '--access-logfile', '-']
elephant = subprocess.Popen(worker_cmd)
mouse = subprocess.Popen(model_cmd)

while not kill_signal:
    time.sleep(1)
    if elephant.poll() is None or mouse.poll() is None: continue
    else:
        worker.send_signal(15)
        model.send_signal(15)

我想知道有没有更优雅的实现方式？如果有更多应用该如何处理？我也尝试过通过应用调用gunicorn并使用multiprocessing来启动这些应用，期望能有如下形式的接口：

from path.to.elephant import app as elephant
from path.to.mouse import app as mouse

app = SomeMiddleware(apps=[elephant, mouse], num_workers=[1, 10])

然后通过命令gunicorn path.to.root:app -w 1启动。

方案1：自定义Gunicorn Worker类（推荐） #

利用Gunicorn的Worker扩展机制，让不同Worker实例加载不同的Flask应用，对外只暴露一个端口，同时精准控制每个应用的Worker数量。

步骤1：编写自定义Worker类

创建worker_manager.py文件，实现根据Worker ID分配对应应用的逻辑：

from gunicorn.workers.sync import SyncWorker
# 导入你的两个Flask应用
from path.to.elephant import app as elephant_app
from path.to.mouse import app as mouse_app

# 定义每个应用的Worker数量配置
APP_WORKER_CONFIG = [
    {"app": elephant_app, "count": 1},
    {"app": mouse_app, "count": 10},
]

class AppSpecificWorker(SyncWorker):
    def load_wsgi(self):
        # 计算当前Worker对应的应用
        cumulative = 0
        for config in APP_WORKER_CONFIG:
            if self.worker_id < cumulative + config["count"]:
                self.wsgi = config["app"]
                break
            cumulative += config["count"]
        else:
            # 默认 fallback 到mouse应用（可选）
            self.wsgi = mouse_app

步骤2：启动Gunicorn

启动时指定总Worker数（1+10=11）和自定义Worker类，绑定到单个端口：

gunicorn --workers 11 --worker-class worker_manager.AppSpecificWorker --bind 0.0.0.0:5000

这样就能实现：

• 对外只有5000一个端口
• 1个Worker进程处理elephant应用，10个Worker进程处理mouse应用
• 无需额外的端口配置或进程管理脚本

方案2：路由分发+内部Unix套接字（适合需要路由隔离的场景） #

如果需要通过不同URL路径区分两个应用（比如/elephant/*和/mouse/*），可以用Werkzeug的分发中间件配合Unix套接字，避免TCP端口占用：

步骤1：编写分发应用dispatcher.py

from werkzeug.middleware.dispatcher import DispatcherMiddleware
from werkzeug.wsgi import WSGIProxyMiddleware
from path.to.elephant import app as elephant_app
from path.to.mouse import app as mouse_app
import multiprocessing
import gunicorn.app.base

# 启动内部应用的函数
def start_internal_app(app, bind, workers):
    class StandaloneApplication(gunicorn.app.base.BaseApplication):
        def __init__(self, app, options=None):
            self.options = options or {}
            self.application = app
            super().__init__()

        def load_config(self):
            config = {key: value for key, value in self.options.items()
                      if key in self.cfg.settings and value is not None}
            for key, value in config.items():
                self.cfg.set(key.lower(), value)

        def load(self):
            return self.application

    StandaloneApplication(app, {
        'bind': bind,
        'workers': workers,
        'accesslog': '-'
    }).run()

# 启动内部应用（用Unix套接字，避免端口冲突）
multiprocessing.Process(target=start_internal_app, args=(elephant_app, 'unix:/tmp/elephant.sock', 1)).start()
multiprocessing.Process(target=start_internal_app, args=(mouse_app, 'unix:/tmp/mouse.sock', 10)).start()

# 前端分发应用
app = DispatcherMiddleware(
    # 默认应用（可选，比如404页面）
    lambda environ, start_response: start_response('404 Not Found', [('Content-Type', 'text/plain')], b'Not Found'),
    {
        '/elephant': WSGIProxyMiddleware('unix:/tmp/elephant.sock'),
        '/mouse': WSGIProxyMiddleware('unix:/tmp/mouse.sock')
    }
)

步骤2：启动分发应用

gunicorn dispatcher:app --bind 0.0.0.0:5000 --workers 1

访问http://localhost:5000/elephant/*会路由到elephant应用，http://localhost:5000/mouse/*路由到mouse应用，内部用Unix套接字通信，无额外TCP端口。

如果有更多应用，只需修改对应的配置：

• 方案1：在APP_WORKER_CONFIG中添加新的应用配置项，比如：

  APP_WORKER_CONFIG = [
      {"app": elephant_app, "count": 1},
      {"app": mouse_app, "count": 10},
      {"app": cat_app, "count": 3},
      {"app": dog_app, "count": 5},
  ]
  

  启动时总Worker数改为1+10+3+5=19即可。

• 方案2：添加新的内部应用启动进程和分发路由规则，比如：

  multiprocessing.Process(target=start_internal_app, args=(cat_app, 'unix:/tmp/cat.sock', 3)).start()
  # ...
  app = DispatcherMiddleware(
      # ...
      {
          '/elephant': WSGIProxyMiddleware('unix:/tmp/elephant.sock'),
          '/mouse': WSGIProxyMiddleware('unix:/tmp/mouse.sock'),
          '/cat': WSGIProxyMiddleware('unix:/tmp/cat.sock'),
      }
  )
  

内容的提问来源于stack exchange，提问作者Akhil Vempali