时间戳

首页 > 解决方案 > 如何在heroku上使用flask实现线程

python - 如何在heroku上使用flask实现线程

问题描述

我有以下代码用于测试在 heroku 上使用烧瓶运行两个线程。

应用程序.py

from flask import Flask, render_template
import threading
import time
import sys

app = Flask(__name__, static_url_path='')
test_result = 'failed'

@app.route('/')
def index():
    return 'Hello! Server is running'


@app.route('/thread-test')
def thread_test():
    global test_result
    return test_result


def thread_testy():
    time.sleep(10)
    global test_result
    test_result = 'passed'
    return


if __name__ == "__main__":
    threading.Thread(target=app.run).start()
    threading.Thread(target=thread_testy).start()

Procile

web: gunicorn app:app --log-file=-

这在本地返回“通过”，但在 heroku 上“失败”。有人对如何让这个测试起作用有任何想法吗？

标签： pythonmultithreadingflaskheroku

解决方案

好的，经过大量的反复试验，我终于找到了解决方案。关键是开始你的新线程@app.before_first_request而不是 in __main__。

应用程序.py

from flask import Flask, render_template
import threading
import time
import sys
app = Flask(__name__, static_url_path='')
test_result = 'failed'

@app.before_first_request
def execute_this():
    threading.Thread(target=thread_testy).start()

@app.route('/')
def index():
    return 'Hello! Server is running successfully'

@app.route('/thread-test')
def thread_test():
    global test_result
    return test_result

def thread_testy():
    time.sleep(10)
    print('Thread is printing to console')
    sys.stdout.flush()
    global test_result
    test_result = 'passed'
    return

def start_app():
    threading.Thread(target=app.run).start()

if __name__ == "__main__":
    start_app()

以上在 10 秒后在 /thread-test 处返回成功

推荐阅读