Python定时器schedule用法

admin管理员组

文章数量:1130349

“Python job scheduling for humans” 是一个名为 schedule 的 Python 库，它提供了一种简单而直观的方式来执行定时任务。这个库的口号是 “Python job scheduling for humans”，意味着它的设计初衷是让任务调度变得更加简单和人性化。

schedule 库的特点包括：

1. 简单易用：schedule 库提供了直观的 API，使得定时任务的安排变得非常简单。

2. 灵活性：你可以根据需要设置不同的时间间隔来执行任务，比如每隔几秒、几分钟、几小时等。

3. 轻量级：schedule 库本身很小巧，不会增加太多额外的负担。

使用 schedule 库的基本步骤如下：

1. 安装 schedule 库：你可以使用 pip 安装 schedule 库，命令为 pip install schedule。

2. 导入 schedule 模块：在你的 Python 脚本中导入 schedule 模块，通常是 import schedule。

3. 定义任务函数：编写你想要定时执行的任务函数。

4. 设置定时任务：使用 schedule.every().[time_unit].do(job) 的形式来设置定时任务，其中 [time_unit] 可以是 seconds、minutes、hours 等，job 是你定义的任务函数。

5. 运行调度器：在一个循环中调用 schedule.run_pending() 方法来运行调度器，以便检查是否有任务需要执行。

示例代码1：

import schedule
import time

def job():
    print("定时任务执行时间：{}".format(time.strftime("%Y-%m-%d %H:%M:%S")))

# 每5秒执行一次任务
schedule.every(5).seconds.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

在这个示例中，我们定义了一个名为 job 的任务函数，并使用 schedule.every(5).seconds.do(job) 来设置每5秒执行一次任务。然后在一个循环中不断调用 schedule.run_pending() 来检查是否有任务需要执行。

示例代码2：flask

在这段代码中，通过使用 threading.Thread 创建了一个新的线程 scheduler_thread，并将 run_scheduler 函数作为目标传递给这个线程。主要目的是为了在 Flask 应用运行的同时，另外开启一个线程来执行定时任务，以避免阻塞应用的主线程。

在 Flask 应用中，如果在主线程中运行一个循环来执行定时任务，可能会导致应用无法响应其他请求，因为主线程会一直被占用。通过将定时任务放在单独的线程中运行，可以确保应用的主线程仍然可以正常处理请求，同时定时任务也能按照预期执行。

因此，使用 threading.Thread 创建一个新的线程来运行定时任务是一种常见的做法，可以确保应用的稳定性和响应性。这样做可以避免定时任务与应用主要功能之间的竞争和阻塞，提高了应用的整体性能。

import time

from flask import Flask
import logging
import pyodbc
import requests
import datetime
import schedule
import threading

app = Flask(__name__)

# 配置日志记录器
# 其他代码部分与你提供的代码保持一致

def job():
    logging.info('定时任务开始时间: {}'.format(datetime.datetime.now()))

    # 在这里插入你的任务代码

    logging.info('定时任务执行完成')
    print('定时任务开始时间: {}'.format(datetime.datetime.now()))

def run_scheduler():
    while True:
        schedule.run_pending()
        time.sleep(1)

# 每5秒钟执行一次任务
schedule.every(5).seconds.do(job)

if __name__ == '__main__':
    scheduler_thread = threading.Thread(target=run_scheduler)
    scheduler_thread.start()
    app.run(debug=True, port=5000)

“Python job scheduling for humans” 是一个名为 schedule 的 Python 库，它提供了一种简单而直观的方式来执行定时任务。这个库的口号是 “Python job scheduling for humans”，意味着它的设计初衷是让任务调度变得更加简单和人性化。

schedule 库的特点包括：

1. 简单易用：schedule 库提供了直观的 API，使得定时任务的安排变得非常简单。

2. 灵活性：你可以根据需要设置不同的时间间隔来执行任务，比如每隔几秒、几分钟、几小时等。

3. 轻量级：schedule 库本身很小巧，不会增加太多额外的负担。

使用 schedule 库的基本步骤如下：

1. 安装 schedule 库：你可以使用 pip 安装 schedule 库，命令为 pip install schedule。

2. 导入 schedule 模块：在你的 Python 脚本中导入 schedule 模块，通常是 import schedule。

3. 定义任务函数：编写你想要定时执行的任务函数。

4. 设置定时任务：使用 schedule.every().[time_unit].do(job) 的形式来设置定时任务，其中 [time_unit] 可以是 seconds、minutes、hours 等，job 是你定义的任务函数。

5. 运行调度器：在一个循环中调用 schedule.run_pending() 方法来运行调度器，以便检查是否有任务需要执行。

示例代码1：

import schedule
import time

def job():
    print("定时任务执行时间：{}".format(time.strftime("%Y-%m-%d %H:%M:%S")))

# 每5秒执行一次任务
schedule.every(5).seconds.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

在这个示例中，我们定义了一个名为 job 的任务函数，并使用 schedule.every(5).seconds.do(job) 来设置每5秒执行一次任务。然后在一个循环中不断调用 schedule.run_pending() 来检查是否有任务需要执行。

示例代码2：flask

在这段代码中，通过使用 threading.Thread 创建了一个新的线程 scheduler_thread，并将 run_scheduler 函数作为目标传递给这个线程。主要目的是为了在 Flask 应用运行的同时，另外开启一个线程来执行定时任务，以避免阻塞应用的主线程。

在 Flask 应用中，如果在主线程中运行一个循环来执行定时任务，可能会导致应用无法响应其他请求，因为主线程会一直被占用。通过将定时任务放在单独的线程中运行，可以确保应用的主线程仍然可以正常处理请求，同时定时任务也能按照预期执行。

因此，使用 threading.Thread 创建一个新的线程来运行定时任务是一种常见的做法，可以确保应用的稳定性和响应性。这样做可以避免定时任务与应用主要功能之间的竞争和阻塞，提高了应用的整体性能。

import time

from flask import Flask
import logging
import pyodbc
import requests
import datetime
import schedule
import threading

app = Flask(__name__)

# 配置日志记录器
# 其他代码部分与你提供的代码保持一致

def job():
    logging.info('定时任务开始时间: {}'.format(datetime.datetime.now()))

    # 在这里插入你的任务代码

    logging.info('定时任务执行完成')
    print('定时任务开始时间: {}'.format(datetime.datetime.now()))

def run_scheduler():
    while True:
        schedule.run_pending()
        time.sleep(1)

# 每5秒钟执行一次任务
schedule.every(5).seconds.do(job)

if __name__ == '__main__':
    scheduler_thread = threading.Thread(target=run_scheduler)
    scheduler_thread.start()
    app.run(debug=True, port=5000)

本文标签： 定时器pythonschedule