无法在一个Python应用程序中同时运行控制台应用程序和API端点

Question

我有一个 python 控制台应用程序，我正在尝试将其容器化在 AWS ECS 中。但要对此应用程序运行健康检查，它需要运行一个 API。

为了解决这个问题，我尝试添加仅返回代码 200“OK”的 API 端点。我希望控制台应用程序和 API 端点同时运行，因此我使用了 ayncio 库。现在，当我运行应用程序时，API 端点快速启动，但控制台应用程序未执行。我觉得 API 端点正在阻止控制台应用程序运行。请帮忙！

app = Flask(__name__)


@app.route('/health', methods=['GET'])
def health_check():
    return "OK", 200


p = Process()

@asyncio.coroutine
def run_console():
    if (p.options.id):
        getattr(p, p.options.action)([int(j) for j in p.options.id.split(',')])

    else:
        getattr(p, p.options.main)()


@asyncio.coroutine
def run_healthcheck_api():
    app.run(port=80, host='0.0.0.0')


loop = asyncio.get_event_loop()

# Run both methods concurrently
loop.run_until_complete(asyncio.gather(run_console(),run_healthcheck_api()))

Answer 1

您似乎正在尝试使用 asyncio 同时运行 Flask API 和控制台应用程序。然而，您面临的问题是 Flask API 启动很快，但控制台应用程序没有执行，可能是由于 API 端点阻止了它。

要解决此问题，您可以使用线程来同时运行 Flask API 和控制台应用程序。这是使用线程的代码的更新版本：

from flask import Flask
from threading import Thread

app = Flask(__name__)

@app.route('/health', methods=['GET'])
def health_check():
    return "OK", 200

class Process:
    # Your Process class implementation

p = Process()

def run_console():
    if (p.options.id):
        getattr(p, p.options.action)([int(j) for j in p.options.id.split(',')])
    else:
        getattr(p, p.options.main)()

def run_healthcheck_api():
    app.run(port=80, host='0.0.0.0')

# Create threads for console application and Flask API
console_thread = Thread(target=run_console)
api_thread = Thread(target=run_healthcheck_api)

# Start both threads
console_thread.start()
api_thread.start()

# Join threads to wait for their completion
console_thread.join()
api_thread.join()

在此代码中，我使用

Thread

模块中的

threading

类为控制台应用程序 (

run_console

) 和 Flask API (

run_healthcheck_api

) 创建单独的线程。两个线程都使用

start()

方法启动，然后使用

join()

方法加入以等待完成。

这种方法应该允许控制台应用程序和 Flask API 同时运行，而不会互相阻塞。请记住将

# Your Process class implementation

替换为实际的

Process

类实现。