服务超时时杀死单个线程答案

【问题标题】：Kill individual threads when timeout in a service服务超时时杀死单个线程
【发布时间】：2019-10-08 13:03:11
【问题描述】：

我正在构建一个 python 烧瓶服务，我正在尝试为每个单独的 POST 请求设置超时。据我了解，每当有人向我的 RESTful 服务发送发布请求时，都会有一个新线程（虚拟或真实）开始执行它。

现在，为了让我的服务器处理大量请求，如果进程运行的时间超过为每个 POST 方法设置的固定时间 (TIMEOUT_TIME)，我希望它返回 TIME-OUT 响应并停止执行那个单独的线程。

你能提出一个我可以使用烧瓶方法实现的抽象方案吗？

【问题讨论】：

【解决方案1】：

一种方法是在单独的进程中运行请求处理并在超时时终止它：

#!/usr/bin/env python3
import time
from multiprocessing import Process

from flask import Flask, request, jsonify


app = Flask(__name__)


@app.route('/api/sleep', methods=['POST'])
def sleep():
    duration = int(request.args.get('duration', 1))
    timeout = float(request.args.get('timeout', 2))

    proc = Process(target=process_request, args=(duration,))
    proc.start()
    proc.join(timeout)

    if proc.is_alive():
        proc.terminate()
        proc.join()

        return jsonify(success=False, message='timeout exceeded'), 408

    return jsonify(success=True, message='well done')


def process_request(t):
    time.sleep(t)


if __name__ == '__main__':
    app.run(host='localhost', port=8080, debug=True)

在此示例中，当睡眠 duration 小于给定的 timeout 时，用户将获得成功响应：

curl -X POST http://localhost:8080/api/sleep?duration=1\&timeout=2
{
  "message": "well done", 
  "success": true
}

否则用户会得到408错误：

curl -X POST http://localhost:8080/api/sleep?duration=2\&timeout=1
{
  "message": "timeout exceeded", 
  "success": false
}

docs 中指出了这种方法的问题

请注意，退出处理程序和 finally 子句等将不会被执行。

这意味着正在运行的进程在退出之前将无法清理，这可能会导致问题。另一种解决方案是使用一个特殊的Joiner 线程，在超时的情况下稍后将用于加入工作进程或线程：

#!/usr/bin/env python3
import time
from queue import Queue
from threading import Thread

from flask import Flask, request, jsonify


class Joiner(Thread):

    def __init__(self):
        super().__init__()
        self.workers = Queue()

    def run(self):

        while True:
            worker = self.workers.get()

            if worker is None:
                break

            worker.join()


app = Flask(__name__)


@app.route('/api/sleep', methods=['POST'])
def sleep():
    duration = int(request.args.get('duration', 1))
    timeout = int(request.args.get('timeout', 2))

    worker = Thread(target=process_request, args=(duration,))
    worker.start()
    worker.join(timeout)

    if worker.is_alive():
        joiner.workers.put(worker)

        return jsonify(success=False, message='timeout exceeded'), 408

    return jsonify(success=True, message='well done')


def process_request(t):
    time.sleep(t)


if __name__ == '__main__':
    joiner = Joiner()
    joiner.start()

    app.run(host='localhost', port=8080, debug=True)

    joiner.workers.put(None)
    joiner.join()

在这里，在运行烧瓶服务器之前，会创建并启动一个 Joiner 线程实例。一旦服务器停止，我们将None 放入joiner.workers 队列以通知joiner 线程完成。

【讨论】：