You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Python Flask中异步执行定时API任务而不阻塞主服务？

avatar
数据老陈
2026-8-2
问题描述

我通过football-data.org API获取赛事数据，为规避请求限制加入了定时逻辑，但将该代码整合到Flask网站主文件后，time.sleep()会阻塞其他核心功能的执行。请问是否有办法实现这些任务与核心功能同时运行而不造成阻塞？

原代码如下：

from flask import Flask, redirect, render_template, flash, request, url_for
from pymongo import MongoClient
import http.client
import json
import time
import requests


app = Flask(__name__)

cluster = MongoClient("mongodb://localhost:27017/footballScoreDB")
db = cluster["footballScorePy"]
collection = db["footballScore"]

        
def get_matches(league):

    status = ["FINISHED", "LIVE", "SCHEDULED"]

    x = 0

    while(x < 3):
        

        uri = "https://api.football-data.org/v4/competitions/" + league + "/matches?status=" + status[x]
        headers = { 'X-Auth-Token': 'api key' }

        response = requests.get(uri, headers=headers)   
        

        matches = response.json()["matches"]

        length = response.json()["resultSet"]["count"]
        i=0


        collection.delete_many({"league": league , "status": status[x]})

        while(i != length): 
        
            awayTeam = matches[i]["awayTeam"]["name"]   
            homeTeam = matches[i]["homeTeam"]["name"]
            homeTeamCrest = matches[i]["homeTeam"]["crest"]
            awayTeamCrest = matches[i]["awayTeam"]["crest"]

            date = matches[i]["utcDate"]
            winner = matches[i]["score"]["winner"]

            awayScore = matches[i]["score"]["fullTime"]["away"]
            awayHalfTimeScore = matches[i]["score"]["halfTime"]["away"]

            homeScore = matches[i]["score"]["fullTime"]["home"]
            homeHalfTimeScore = matches[i]["score"]["halfTime"]["home"]
    
            if status == "FINISHED":

                post = {
                    "homeTeam": homeTeam,
                    "awayTeam": awayTeam,
                    "date": date,
                    "homeScore": homeScore,
                    "awayScore": awayScore,
                    "homeHalfTimeScore": homeHalfTimeScore,
                    "awayHalfTimeScore": awayHalfTimeScore,
                    "winner": winner,
                    "status": status[x],
                    "league": league,
                    "awayTeamCrest": awayTeamCrest,
                    "homeTeamCrest": homeTeamCrest
                }

            elif status == "LIVE":  

                    post = {
                        "homeTeam": homeTeam,
                        "awayTeam": awayTeam,
                        "date": "LIVE",
                        "homeScore": homeScore,
                        "awayScore": awayScore,
                        "homeHalfTimeScore": 0,
                        "awayHalfTimeScore": 0,
                        "winner": 0,
                        "status": status[x],
                        "league": league,
                        "awayTeamCrest": awayTeamCrest,
                        "homeTeamCrest": homeTeamCrest
                    }

            else :

                    post = {
                        "homeTeam": homeTeam,
                        "awayTeam": awayTeam,
                        "date": date,
                        "homeScore": 0,
                        "awayScore": 0,
                        "homeHalfTimeScore": 0,
                        "awayHalfTimeScore": 0,
                        "winner": 0,
                        "status": status[x],
                        "league": league,
                        "awayTeamCrest": awayTeamCrest,
                        "homeTeamCrest": homeTeamCrest
                    }
                    
            if collection.find_one(post) == False :
                collection.insert_one(post)
                
            
            i+=1

        x+=1   


@app.route("/")
def home():

    return "testing"



if __name__ == "__main__":
    app.run(debug=True, port="4000")
    while True:
        get_matches("PL")
        get_matches("SA")
        get_matches("DED")
        time.sleep(60)
解决方案

方法1：使用Python内置threading模块

将定时任务放到独立线程中运行，避免阻塞Flask主进程。同时修复原代码中的两处逻辑错误：

  1. 状态判断时误用整个status列表，应改为当前循环的status[x]
  2. 数据库存在性判断错误，find_one返回None而非False

修改后的代码：

from flask import Flask, redirect, render_template, flash, request, url_for
from pymongo import MongoClient
import requests
import time
import threading


app = Flask(__name__)

cluster = MongoClient("mongodb://localhost:27017/footballScoreDB")
db = cluster["footballScorePy"]
collection = db["footballScore"]


def get_matches(league):
    status = ["FINISHED", "LIVE", "SCHEDULED"]
    x = 0
    while x < 3:
        uri = f"https://api.football-data.org/v4/competitions/{league}/matches?status={status[x]}"
        headers = {'X-Auth-Token': 'api key'}

        response = requests.get(uri, headers=headers)
        matches = response.json()["matches"]
        length = response.json()["resultSet"]["count"]
        i = 0

        collection.delete_many({"league": league, "status": status[x]})

        while i != length:
            awayTeam = matches[i]["awayTeam"]["name"]
            homeTeam = matches[i]["homeTeam"]["name"]
            homeTeamCrest = matches[i]["homeTeam"]["crest"]
            awayTeamCrest = matches[i]["awayTeam"]["crest"]

            date = matches[i]["utcDate"]
            winner = matches[i]["score"]["winner"]

            awayScore = matches[i]["score"]["fullTime"]["away"]
            awayHalfTimeScore = matches[i]["score"]["halfTime"]["away"]

            homeScore = matches[i]["score"]["fullTime"]["home"]
            homeHalfTimeScore = matches[i]["score"]["halfTime"]["home"]

            if status[x] == "FINISHED":
                post = {
                    "homeTeam": homeTeam,
                    "awayTeam": awayTeam,
                    "date": date,
                    "homeScore": homeScore,
                    "awayScore": awayScore,
                    "homeHalfTimeScore": homeHalfTimeScore,
                    "awayHalfTimeScore": awayHalfTimeScore,
                    "winner": winner,
                    "status": status[x],
                    "league": league,
                    "awayTeamCrest": awayTeamCrest,
                    "homeTeamCrest": homeTeamCrest
                }
            elif status[x] == "LIVE":
                post = {
                    "homeTeam": homeTeam,
                    "awayTeam": awayTeam,
                    "date": "LIVE",
                    "homeScore": homeScore,
                    "awayScore": awayScore,
                    "homeHalfTimeScore": 0,
                    "awayHalfTimeScore": 0,
                    "winner": 0,
                    "status": status[x],
                    "league": league,
                    "awayTeamCrest": awayTeamCrest,
                    "homeTeamCrest": homeTeamCrest
                }
            else:
                post = {
                    "homeTeam": homeTeam,
                    "awayTeam": awayTeam,
                    "date": date,
                    "homeScore": 0,
                    "awayScore": 0,
                    "homeHalfTimeScore": 0,
                    "awayHalfTimeScore": 0,
                    "winner": 0,
                    "status": status[x],
                    "league": league,
                    "awayTeamCrest": awayTeamCrest,
                    "homeTeamCrest": homeTeamCrest
                }

            if not collection.find_one(post):
                collection.insert_one(post)

            i += 1
        x += 1


def scheduled_task():
    while True:
        get_matches("PL")
        get_matches("SA")
        get_matches("DED")
        time.sleep(60)


@app.route("/")
def home():
    return "testing"


if __name__ == "__main__":
    # 启动定时任务线程，daemon=True表示随主进程退出而终止
    task_thread = threading.Thread(target=scheduled_task, daemon=True)
    task_thread.start()
    # 启动Flask服务
    app.run(debug=True, port="4000")

方法2：使用APScheduler扩展（适合复杂定时需求）

如果需要更灵活的定时配置（如Cron表达式、指定时间点执行），可以使用APScheduler这个Flask常用的定时任务扩展。

步骤1：安装依赖 

pip install apscheduler

步骤2：修改代码整合APScheduler 

from flask import Flask
from pymongo import MongoClient
import requests
from apscheduler.schedulers.background import BackgroundScheduler


app = Flask(__name__)

cluster = MongoClient("mongodb://localhost:27017/footballScoreDB")
db = cluster["footballScorePy"]
collection = db["footballScore"]


def get_matches(league):
    # 内容同方法1中修正后的get_matches函数，此处省略重复代码
    status = ["FINISHED", "LIVE", "SCHEDULED"]
    x = 0
    while x < 3:
        uri = f"https://api.football-data.org/v4/competitions/{league}/matches?status={status[x]}"
        headers = {'X-Auth-Token': 'api key'}
        response = requests.get(uri, headers=headers)
        matches = response.json()["matches"]
        length = response.json()["resultSet"]["count"]
        i = 0
        collection.delete_many({"league": league, "status": status[x]})
        while i != length:
            awayTeam = matches[i]["awayTeam"]["name"]
            homeTeam = matches[i]["homeTeam"]["name"]
            homeTeamCrest = matches[i]["homeTeam"]["crest"]
            awayTeamCrest = matches[i]["awayTeam"]["crest"]
            date = matches[i]["utcDate"]
            winner = matches[i]["score"]["winner"]
            awayScore = matches[i]["score"]["fullTime"]["away"]
            awayHalfTimeScore = matches[i]["score"]["halfTime"]["away"]
            homeScore = matches[i]["score"]["fullTime"]["home"]
            homeHalfTimeScore = matches[i]["score"]["halfTime"]["home"]
            if status[x] == "FINISHED":
                post = {
                    "homeTeam": homeTeam,
                    "awayTeam": awayTeam,
                    "date": date,
                    "homeScore": homeScore,
                    "awayScore": awayScore,
                    "homeHalfTimeScore": homeHalfTimeScore,
                    "awayHalfTimeScore": awayHalfTimeScore,
                    "winner": winner,
                    "status": status[x],
                    "league": league,
                    "awayTeamCrest": awayTeamCrest,
                    "homeTeamCrest": homeTeamCrest
                }
            elif status[x] == "LIVE":
                post = {
                    "homeTeam": homeTeam,
                    "awayTeam": awayTeam,
                    "date": "LIVE",
                    "homeScore": homeScore,
                    "awayScore": awayScore,
                    "homeHalfTimeScore": 0,
                    "awayHalfTimeScore": 0,
                    "winner": 0,
                    "status": status[x],
                    "league": league,
                    "awayTeamCrest": awayTeamCrest,
                    "homeTeamCrest": homeTeamCrest
                }
            else:
                post = {
                    "homeTeam": homeTeam,
                    "awayTeam": awayTeam,
                    "date": date,
                    "homeScore": 0,
                    "awayScore": 0,
                    "homeHalfTimeScore": 0,
                    "awayHalfTimeScore": 0,
                    "winner": 0,
                    "status": status[x],
                    "league": league,
                    "awayTeamCrest": awayTeamCrest,
                    "homeTeamCrest": homeTeamCrest
                }
            if not collection.find_one(post):
                collection.insert_one(post)
            i += 1
        x += 1


def scheduled_task():
    get_matches("PL")
    get_matches("SA")
    get_matches("DED")


@app.route("/")
def home():
    return "testing"


if __name__ == "__main__":
    # 初始化后台调度器
    scheduler = BackgroundScheduler()
    # 添加定时任务：每60秒执行一次scheduled_task函数
    scheduler.add_job(scheduled_task, 'interval', seconds=60)
    # 启动调度器
    scheduler.start()
    # 启动Flask服务
    app.run(debug=True, port="4000")
注意事项
  1. 生产环境中不要使用app.run()启动Flask，应使用Gunicorn、uWSGI等生产级服务器，同时确保定时任务的启动方式与服务器兼容。
  2. PyMongo客户端本身是线程安全的，无需额外处理数据库操作的线程同步问题。
  3. 严格遵守football-data.org的API请求限制，避免因请求过于频繁被封禁。

内容的提问来源于stack exchange，提问作者user16462824

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
火山引擎 最新活动
全面开放
Doubao-Seedance-2.5全新模型升级

模型再升级,30秒超长叙事, 模态参考扩容

HOT
方舟Codingplan

模型自由，工具不限，最新支持 Deepseek-V4 系列、GLM-5.3 系列

HOT
方舟Agent plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列

HOT
新用户特惠专场

大模型19元起，Al应用9.9元畅享，新人首购爆款尽享优惠

最近更新时间：2026.08.02 16:35:23