在 WSGI/apache 应用程序中强制每个 IP 的并发线程限制

Question

我们正在运行一个 Flask 应用程序来公开存储在数据库中的数据。它返回很多503 错误。我的理解是，当达到最大并发线程数时，这些是由apache生成的。

根本原因很可能是应用程序性能不佳，但在这个阶段，我们无法承受更多的开发时间，因此我正在寻找一种廉价的部署配置黑客来缓解这个问题。

• 数据提供者正在高速发送数据。我相信他们的程序会得到很多 503 并且只需尝试/捕获那些重试直到成功。

• 数据消费者使用该应用的频率要低得多，我希望他们不要被这些问题所困扰。

我正在考虑限制来自每个提供商的 IP 的并发访问数。他们可能会获得较低的吞吐量，但他们会像现在一样接受它，这将使普通消费者的生活更轻松。

---

我发现mod_limitipconn 似乎是为此量身定做的。

mod_limitipconn [...] 允许管理员限制允许来自单个 IP 地址的同时请求数。

我想确定我了解它的工作原理以及限制是如何设置的。

由于 WSGI 设置，我一直认为最多有 5 个同时连接：threads=5。但是我在 mod_wsgi 文档中阅读了Processes and Threading，我很困惑。

考虑到下面的配置，这些假设是否正确？

• 一次只运行一个应用程序实例。

• 最多可以生成 5 个并发线程。

• 当处理 5 个请求时，如果有第六个请求到达，客户端会收到一个503。

• 限制同时请求 IP x.x.x.x 的数量。将 apache 级别设置为 3 将确保该 IP 可以使用这 5 个线程中的 3 个，而将 2 个留给其他 IP。

• 增加 WSGI 配置中的线程数可以通过在速率限制中提供更精细的粒度来帮助在客户端之间共享连接池（您可以将 4 个提供者中的每一个限制为 3 个，并保留 5 个以上，总共 17 个) 但不会提高整体性能，即使服务器有空闲内核，因为the Python GIL prevents several threads to run at the same time。

• 将线程数提高到 100 等高数可能会使请求更长，但会限制 503 响应。如果客户将他们自己的并发请求限制设置得不太高，甚至可能就足够了，如果他们不这样做，我可以使用mod_limitipconn 之类的东西来强制执行。

• 过多地增加线程数会使请求时间过长，以至于客户端会超时，而不是503，这并不是更好。

---

当前配置如下。不知道什么重要。

apachectl -V:

Server version: Apache/2.4.25 (Debian)
Server built:   2018-06-02T08:01:13
Server's Module Magic Number: 20120211:68
Server loaded:  APR 1.5.2, APR-UTIL 1.5.4
Compiled using: APR 1.5.2, APR-UTIL 1.5.4
Architecture:   64-bit
Server MPM:     event
  threaded:     yes (fixed thread count)
    forked:     yes (variable process count)

/etc/apache2/apache2.conf:

# KeepAlive: Whether or not to allow persistent connections (more than
# one request per connection). Set to "Off" to deactivate.
#
KeepAlive On

#
# MaxKeepAliveRequests: The maximum number of requests to allow
# during a persistent connection. Set to 0 to allow an unlimited amount.
# We recommend you leave this number high, for maximum performance.
#
MaxKeepAliveRequests 100

/etc/apache2/mods-available/mpm_worker.conf（但这在event 中应该不重要，对吧？）：

<IfModule mpm_worker_module>
        StartServers                     2
        MinSpareThreads          25
        MaxSpareThreads          75
        ThreadLimit                      64
        ThreadsPerChild          25
        MaxRequestWorkers         150
        MaxConnectionsPerChild   0
</IfModule>

/etc/apache2/sites-available/my_app.conf:

WSGIDaemonProcess my_app threads=5

Answer 1

我希望他们不要被打扰，因此将数据提供者的请求与数据消费者分开（我不熟悉 apache，所以我不会向您展示生产就绪配置，而是总体方法）：

<VirtualHost *>
    ServerName example.com

    WSGIDaemonProcess consumers user=user1 group=group1 threads=5
    WSGIDaemonProcess providers user=user1 group=group1 threads=5
    WSGIScriptAliasMatch ^/consumers_ulrs/.* /path_to_your_app/consumers.wsgi process-group=consumers
    WSGIScriptAliasMatch ^/providers_ulrs/.* /path_to_your_app/providers.wsgi process-group=providers

    ...

</VirtualHost>

通过限制每个 IP 的请求数量，您可能会损害用户体验，但仍然无法解决您的问题。例如，请注意，由于 NAT 和 ISP 的工作方式，许多独立用户可能拥有相同的 IP。

附： ThreadsPerChild=25 但WSGIDaemonProcess my_app threads=5 很奇怪。您确定通过该配置，Apache 创建的所有线程都会被 WSGI 服务器使用吗？

Answer 2

我最终采用了不同的方法。我在应用程序代码中添加了一个限制器来处理这个问题。

"""Concurrency requests limiter

Inspired by Flask-Limiter
"""

from collections import defaultdict
from threading import BoundedSemaphore
from functools import wraps

from flask import request
from werkzeug.exceptions import TooManyRequests


# From flask-limiter
def get_remote_address():
    """Get IP address for the current request (or 127.0.0.1 if none found)

    This won't work behind a proxy. See flask-limiter docs.
    """
    return request.remote_addr or '127.0.0.1'


class NonBlockingBoundedSemaphore(BoundedSemaphore):
    def __enter__(self):
        ret = self.acquire(blocking=False)
        if ret is False:
            raise TooManyRequests(
                'Only {} concurrent request(s) allowed'
                .format(self._initial_value))
        return ret


class ConcurrencyLimiter:

    def __init__(self, app=None, key_func=get_remote_address):
        self.app = app
        self.key_func = key_func
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        self.app = app
        app.extensions = getattr(app, 'extensions', {})
        app.extensions['concurrency_limiter'] = {
            'semaphores': defaultdict(dict),
        }

    def limit(self, max_concurrent_requests=1):
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # Limiter not initialized
                if self.app is None:
                    return func(*args, **kwargs)
                identity = self.key_func()
                sema = self.app.extensions['concurrency_limiter'][
                    'semaphores'][func].setdefault(
                        identity,
                        NonBlockingBoundedSemaphore(max_concurrent_requests)
                    )
                with sema:
                    return func(*args, **kwargs)
            return wrapper
        return decorator


limiter = ConcurrencyLimiter()


def init_app(app):
    """Initialize limiter"""

    limiter.init_app(app)
    if app.config['AUTHENTICATION_ENABLED']:
        from h2g_platform_core.api.extensions.auth import get_identity
        limiter.key_func = get_identity

然后我需要做的就是将该装饰器应用到我的视图中：

@limiter.limit(1)  # One concurrent request by user
def get(...):
    ...

实际上，我只保护产生高流量的那些。

在应用程序代码中这样做很好，因为我可以为每个经过身份验证的用户而不是每个 IP 设置一个限制。

为此，我需要做的就是将key_func 中的默认get_remote_address 替换为返回用户唯一标识的函数。

请注意，这为每个视图函数设置了不同的限制。如果限制需要是全局的，则可以以不同的方式实现。事实上，它会更简单。