Flask 中怎样用 while 写一个定时执行的任务？（定时查询最新数据）

[miniyao]

用 before_app_first_request 的钩子，希望 app 首次启动时，开始运行一个每隔 1 小时定时查询最新数据的任务。结果发现这个任务每隔 1 小时去查询最新的数据，通过 flask_sqlalchemy 初始化之后的工厂模式，查询到的数据都是一样的。如下：

def async_cron_task(app):

with app.app_context():

while True:

new_user = User.query.order_by(User.reg_time.desc()).first()

print('New User: ', new_user.id)

time.sleep(3600)



def cron_task():

app = current_app._get_current_object()

thr = Thread(target=async_cron_task, args=[app])

thr.start()

return thr

但是，如果把数据库写成硬连接，就可以查到最新数据：

db = MySQLdb.connect(host='localhost', port=3306, user='xxx', passwd='xxx', db='xxx', charset='utf8')

cursor = db.cursor()

由于原生 SQL 查询语句太复杂了，希望用 SQLAlchemy 的方式连接数据库，要怎么样才能定时查到新数据呢？

[cz5424]

Flask 不适合做定时，定时任务可以选择系统 crontab 执行脚本或者 celery

[wellsc]

我没看懂，写个异步任务脚本？

[BingoXuan]

轻量就用 huey 做异步队列

[miniyao]

@cz5424 后面我提到了，因为要写非常复杂的 sql 去查询数据、处理完了还要写回系统，系统 crontab 需要单独写不少代码。celery 用过，过了几个星期自动假死，不敢用了，没有 app 级的 while 可靠。

@wellsc 对，可以这么理解。异步任务中，写了一个死循环（定时任务）。

[miniyao]

@BingoXuan 就是不想用队列工具，增加一层不确定性。毕竟 app 级 while 很靠谱呀

[cz5424]

celery 并不是玩具，只是你不熟悉，实际上给的这些看不出你查询每次都一样的问题在哪里，可以尝试一下调换这两个的顺序

```python
with app.app_context():

while True:
```

[miniyao]

@cz5424 每次查询的数据都一样的意思是，就是无论系统有没有新赠的 new_user.id ，这里面 print 的 id 都不会变。感谢！上下调换了一下顺序，数据就可以更新了，应该是上下文没有切换。

[MintZX]

对啊，你这个明显是在同样的 context 里面空转啊。。。可不是每次获得的数据一样

[knightdf]

@miniyao celery 用过，过了几个星期自动假死，不敢用了，没有 app 级的 while 可靠。
自己写的有问题吧？我们线上产品都用了 celery，毛事没有

[liuxingdeyu]

我的思路是，定时任务、api 、各种 handler 自己干自己的，然后从启动命令里区分出来。如果需要通信就搞个消息通道。之前我们的项目就是一大坨面条似的东西，又弱又乱，后来我拆分出来 dao 层，把各种工具类还有枚举统一放，把 api 和定时任务拆开，各干各的。这样扩容也方便，自己查自己改也方便，容器化之后还能搞一堆镜像各用各的。flask 做定时任务除了 celery 我记得还有个 apscheduler，不过这东西贼坑，只能用线程（ mokey_pach 后会切换失败）

[ytmsdy]

用 crontab，或者 celery 吧。
自己写太费劲了

[yingxiangyu]

celery+apscheduler

[yingxiangyu]

celery 定时任务不太好使，定时可以用 apscheduler 实现，触发任务 celery 异步执行

[ila]

用 mysql 的 event schedule

[Hardrain]

要定时执行的单独一个文件，建一个 oneshot 的 systemd service 配合 timer 就好了
cron 过时了

[sunhk25]

@miniyao celery 自动假死是指定期任务不执行吗？

[hunk]

Flask+ apscheduler

[ch2]

apscheduler 插件

[triptipstop]

crontab 怎么会多写代码？
http 接口写好，定时 get 一下就行。

[kkx]

crontab 需要多写很多代码说明你代码写的耦合度高。。。

[userff]

apscheduler

[fansfans]

我前几天刚好也遇到了类似的问题在接口中提交数据但是在线程池中的任务没有获取到对应的数据 应该是 sqlalchemy 的缓存问题 后面直接用 celery 了 调式控制都方便很多。

[frostming]

https://github.com/frostming/flask-crontab

用 crontab 运行周期任务，自动激活 app_context，减少代码量

[wobuhuicode]

定时任务用 crontab ～

[RRRoger]

可以试试改成自己调自己的接口

[amoyiki]

目前项目使用很多年 celery, 出现过屈指可数的程序问题。一般就是 redis 获取连接池失败。
不使用 celery 的话可以自己创建一个接口，外部 1h 调用一次。外部直接用 crontab 写定时

[luckyc]

用 flask-apscheduler

[www5070504]

丢到 apscheduler 里

[wzwwzw]

apscheduler flask 有插件

[Vegetable]

对自己好一点，apscheduler 吧

[676529483]

1. 你这个就是数据库连接处理有问题额，问题在 with app.app_context():这行，你应该想要使用 sqlalchemy 的连接初始化，要手动调用获取数据库连接
2. 我以前也这样自己处理，发现各种调度时间问题。现在更倾向于另启动一个 cron 服务，定时去调用 api 接口

[nonduality]

针对你的需求，要获取最新数据，用相关事件信号触发动作不是更直截了当还稳定可靠？

[cz5424]

看了一下的评论，猜测是事务的问题，同一个上下文里面会一直处于同一个事务，可重复读会一直读到相同数据，另一个方法是可以修改 sqlalchemy 的事务隔离级别来避免，不需要调换上下文和 while （代价是其他读写提交也会生效，慎重修改）

[HUNYXV]

```
pip install apscheduler
```

[jianhua]

搜索：flask + celery

[Aether]

@knightdf 一直在用 celery，没出过问题（好几年了）。你多研究下？

[knightdf]

@Aether 你回错人了？

[myCupOfTea]

apscheduler 有的时候会忽略任务,比如当前系统比较繁忙正好到了定时任务的时候

[fansfans]

@cz5424 我目前复现情况也听多的 技术在数据库中查询有数据 使用 sqlalchemy 依旧无法获取最新数据 。

[yagamil]

看着像脏读的问题. 每次用 sql 连接语句重连就不出现这个问题

[fansfans]

其实这个和 celery 、apscheduler 好像没什么关联、应该是 session 的上下文使用的是同一个、导致的脏读、但是在接口中进行了 commit 不应该出现这种情况才对、也许可以通过自己创建 session 解决 db_session = sessionmaker(bind=db.engine)()

[nicolaz]

```python
def set_interval(interval):
def decorator(function):
def wrapper(*args, **kwargs):
stopped = threading.Event()

def loop(): # executed in another thread
while not stopped.wait(interval): # until stopped
function(*args, **kwargs)

t = threading.Thread(target=loop)
t.daemon = True # stop if the program exits
t.start()
return stopped

return wrapper

return decorator


@set_interval(60*60)
def do_something():
pass
```

[seven123]

celery 的确有时候会出现一些诡异问题，直接用 apscheduler 把