flask 异步任务

from flask import Flask
from flask_apscheduler import APScheduler

app = Flask(__name__)

class Config:
    SCHEDULER_API_ENABLED = True

app.config.from_object(Config())

scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()

@app.route('/')
def index():
    return "Hello, Flask!"

@scheduler.task('interval', id='do_job_1', seconds=30)
def job1():
    print("Job 1 executed!")

if __name__ == '__main__':
    app.run()

pip install celery redis

from flask import Flask
from celery import Celery

app = Flask(__name__)

app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@app.route('/')
def index():
    return "Hello, Flask!"

@celery.task
def add_together(a, b):
    return a + b

if __name__ == '__main__':
    app.run()

from your_flask_app import add_together

result = add_together.delay(23, 42)
print(result.get())  # 输出 65