Python实现定时任务的方式

        大家好,在当今数字化的时代,定时任务的需求在各种应用场景中频繁出现。无论是数据的定时更新、周期性的任务执行,还是特定时间点的操作触发,Python 都为我们提供了强大而灵活的手段来实现这些定时任务。当我们深入探索 Python 的世界时,会惊喜地发现多种巧妙的方式能够让我们精准地掌控时间,达成各种定时需求。在接下来的文章中,我们将一同揭开 Python 定时任务实现方式的神秘面纱,探寻其背后的精彩与奥秘。

一、使用time.sleep()函数

        使用time.sleep()函数是一种简单的实现定时任务的方式,尽管它并不是真正意义上的定时任务。该函数可以让程序在执行到该语句时暂停一段时间,从而实现在指定的时间间隔后执行下一步操作。但需要注意的是,time.sleep()会阻塞程序的执行,意味着在等待的过程中,程序无法同时执行其他任务,这在一些情况下可能会造成问题。

下面是一个简单的示例,演示如何使用time.sleep()函数实现定时任务:

import time

def my_task():
    print("定时任务执行中...")
    # 这里放置你想要定时执行的任务代码

def main():
    while True:
        my_task()
        # 暂停10秒钟后再次执行任务
        time.sleep(10)

if __name__ == "__main__":
    main()

        在这个示例中,my_task()函数定义了我们想要定时执行的任务。在main()函数中,通过一个无限循环来反复执行my_task()函数,并在每次执行后暂停10秒钟。这样就实现了每隔10秒执行一次任务的效果。

        然而,需要注意的是,由于time.sleep()会阻塞程序的执行,因此在这个例子中,除了定时任务外,程序不会执行任何其他操作。如果需要同时执行其他任务,就需要考虑使用更为灵活的定时任务库,如scheduleAPScheduler

二、使用threading.Timer

        使用threading.Timer类是一种更加灵活的实现定时任务的方式,它允许你在指定的时间后执行函数,而且不会阻塞程序的执行。通过创建一个Timer对象,你可以设置定时器的间隔时间和要执行的函数,然后启动定时器,这样就可以实现定时执行任务的效果。

下面是一个示例,演示如何使用threading.Timer类实现定时任务:

import threading

def my_task():
    print("定时任务执行中...")
    # 这里放置你想要定时执行的任务代码

def start_timer(interval):
    # 创建一个定时器对象,设置定时器的间隔时间和要执行的函数
    timer = threading.Timer(interval, my_task)
    # 启动定时器
    timer.start()

if __name__ == "__main__":
    # 设置定时器的间隔时间(单位:秒)
    interval = 10
    # 启动定时器
    start_timer(interval)

        在这个示例中,我们首先定义了一个my_task()函数,它包含了我们想要定时执行的任务代码。然后定义了一个start_timer()函数,用来创建和启动定时器。在start_timer()函数中,我们通过threading.Timer(interval, my_task)创建了一个定时器对象,指定了定时器的间隔时间为interval秒,并设置要执行的函数为my_task()。最后通过timer.start()启动定时器。

        这样,每隔interval秒,定时器就会执行一次my_task()函数,从而实现了定时执行任务的效果。与使用time.sleep()不同的是,使用threading.Timer类实现定时任务不会阻塞程序的执行,因此可以同时执行其他任务。

三、使用sched模块

        使用Python标准库中的sched模块是实现定时任务的另一种常见方式。sched模块提供了一个事件调度器,可以用于调度在将来某个时间执行的任务。通过使用sched.scheduler类,你可以创建一个调度器对象,并使用enter()方法安排任务的执行。

下面是一个示例,演示如何使用sched模块实现定时任务:

import sched
import time

# 创建一个调度器对象
scheduler = sched.scheduler(time.time, time.sleep)

def my_task():
    print("定时任务执行中...")
    # 这里放置你想要定时执行的任务代码

def schedule_task(interval):
    # 设置任务的执行时间
    next_run_time = time.time() + interval
    # 安排任务的执行
    scheduler.enterabs(next_run_time, 1, my_task)
    print("定时任务已安排,下次执行时间为:", time.ctime(next_run_time))

if __name__ == "__main__":
    # 设置定时任务的间隔时间(单位:秒)
    interval = 10
    # 安排第一次执行任务
    schedule_task(interval)
    # 开始调度器,持续执行已安排的任务
    scheduler.run()

        在这个示例中,我们首先导入了sched模块和time模块。然后创建了一个调度器对象scheduler,使用了sched.scheduler(time.time, time.sleep)来初始化调度器对象,指定了时间函数和睡眠函数。接着定义了一个my_task()函数,其中包含了我们想要定时执行的任务代码。然后定义了一个schedule_task()函数,用于安排任务的执行。在schedule_task()函数中,我们通过scheduler.enterabs(next_run_time, 1, my_task)来安排任务的执行,指定了任务的执行时间、优先级和要执行的函数。最后,在if __name__ == "__main__":中,我们设置了定时任务的间隔时间为interval秒,并安排了第一次执行任务。然后通过scheduler.run()开始调度器,使其持续执行已安排的任务。

这样,每隔interval秒,调度器就会执行一次my_task()函数,从而实现了定时执行任务的效果。

四、使用asyncio库

        使用asyncio库是在基于异步编程的项目中实现定时任务的一种有效方式。asyncio库提供了异步编程的支持,允许你在程序中定义协程(coroutines),并使用async/await关键字来管理异步任务的执行。

        当需要在一段时间后执行某个任务时,你可以使用asyncio.sleep()函数创建一个异步等待,然后在等待结束后执行相应的任务。

下面是一个示例,演示如何使用asyncio库实现定时任务:

import asyncio

async def my_task():
    print("定时任务执行中...")
    # 这里放置你想要定时执行的任务代码

async def schedule_task(interval):
    while True:
        # 等待一段时间后执行任务
        await asyncio.sleep(interval)
        # 执行任务
        await my_task()

if __name__ == "__main__":
    # 设置定时任务的间隔时间(单位:秒)
    interval = 10
    # 创建事件循环对象
    loop = asyncio.get_event_loop()
    # 启动定时任务
    loop.create_task(schedule_task(interval))
    # 运行事件循环
    loop.run_forever()

        在这个示例中,我们首先导入了asyncio库。然后定义了一个my_task()协程函数,其中包含了我们想要定时执行的任务代码。接着定义了一个schedule_task()协程函数,用于实现定时任务的调度。在schedule_task()函数中,我们使用asyncio.sleep(interval)创建了一个异步等待,等待一段时间后执行任务。然后通过await my_task()执行任务。最后,在if __name__ == "__main__":中,我们设置了定时任务的间隔时间为interval秒,并通过loop.create_task(schedule_task(interval))启动定时任务。然后通过loop.run_forever()运行事件循环,使定时任务持续执行。

        这样,每隔interval秒,就会执行一次定时任务。由于使用了asyncio库,定时任务的执行是非阻塞的,因此可以与其他异步任务一起运行,提高了程序的并发性能。

五、使用schedule库

   schedule库提供了一个简单的API,可以方便地安排任务的执行。可以使用schedule.every().xxx方法来设置定时任务的执行间隔和要执行的任务函数,然后调用schedule.run_pending()方法来运行已安排的任务。

下面是一个示例,演示如何使用schedule库实现定时任务:

import schedule
import time

def my_task():
    print("定时任务执行中...")
    # 这里放置你想要定时执行的任务代码

if __name__ == "__main__":
    # 设置定时任务的间隔时间(单位:秒)
    interval = 10
    # 安排定时任务的执行
    schedule.every(interval).seconds.do(my_task)
    
    while True:
        # 运行已安排的任务
        schedule.run_pending()
        time.sleep(1)

        在这个示例中,我们首先导入了schedule库和time模块。然后定义了一个my_task()函数,其中包含了我们想要定时执行的任务代码。接着在if __name__ == "__main__":中,我们设置了定时任务的间隔时间为interval秒,并使用schedule.every(interval).seconds.do(my_task)安排了定时任务的执行。最后通过schedule.run_pending()运行已安排的任务。

六、使用APScheduler库

    APScheduler(Advanced Python Scheduler)库提供了更多高级功能,如支持不同的触发器(包括时间间隔触发器、定时触发器等)、持久化存储等。

下面是一个示例,演示如何使用APScheduler库实现定时任务:

from apscheduler.schedulers.background import BackgroundScheduler

def my_task():
    print("定时任务执行中...")
    # 这里放置你想要定时执行的任务代码

if __name__ == "__main__":
    # 创建一个后台调度器
    scheduler = BackgroundScheduler()
    # 添加定时任务,设置定时触发器,间隔时间为10秒
    scheduler.add_job(my_task, 'interval', seconds=10)
    # 启动调度器
    scheduler.start()
    
    try:
        # 防止主线程结束
        while True:
            pass
    except (KeyboardInterrupt, SystemExit):
        # 停止调度器
        scheduler.shutdown()

        在这个示例中,我们首先导入了BackgroundScheduler类。然后定义了一个my_task()函数,其中包含了我们想要定时执行的任务代码。接着在if __name__ == "__main__":中,我们创建了一个后台调度器scheduler,并使用scheduler.add_job(my_task, 'interval', seconds=10)添加了定时任务,设置了定时触发器,间隔时间为10秒。最后通过scheduler.start()启动调度器,并在程序结束前调用scheduler.shutdown()停止调度器。这样,定时任务就会在每隔10秒执行一次。

七、使用操作系统的调度工具

        使用操作系统的调度工具,如cron(Linux/Unix系统)或Task Scheduler(Windows系统),是一种常见且有效的方式来执行定时任务。通过编写一个Python脚本,并在操作系统的调度工具中设置定时执行的时间和执行命令,可以轻松地实现定时任务的调度和管理。

1、使用cron(Linux系统):

        cron是一个在Unix类操作系统中常用的计划任务程序,它允许用户在预定时间执行命令或脚本。你可以通过编辑crontab文件来设置定时执行的任务。

下面是一个示例,演示如何使用cron来执行Python脚本:

首先,编写一个Python脚本(例如my_script.py):

# my_script.py
def my_task():
    print("定时任务执行中...")
    # 这里放置你想要定时执行的任务代码

if __name__ == "__main__":
    my_task()

然后,在终端中编辑crontab文件,添加定时执行的任务:

$ crontab -e

在打开的编辑器中添加以下内容:

# 每天上午8点执行Python脚本
0 8 * * * /usr/bin/python3 /path/to/my_script.py

保存并关闭编辑器。这样,Python脚本my_script.py就会每天上午8点执行一次。

2、使用Task Scheduler计划任务(Windows系统):

        Task Scheduler是Windows系统中的一个任务计划程序,可以让你安排执行各种任务,包括运行程序、发送电子邮件等。

下面是一个示例,演示如何使用Task Scheduler来执行Python脚本:

首先,编写一个Python脚本(例如my_script.py):

# my_script.py
def my_task():
    print("定时任务执行中...")
    # 这里放置你想要定时执行的任务代码

if __name__ == "__main__":
    my_task()

        然后,打开Task Scheduler并创建一个新的基本任务。在任务设置中,指定要运行的程序为Python解释器,并将Python脚本的路径作为参数传递给解释器。在触发器设置中,选择适当的触发器类型和执行时间。最后,设置完成后保存任务。这样,Python脚本my_script.py就会在指定的时间执行。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/663748.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

代理 模式

一、什么是代理模式 代理模式指代理控制对其他对象的访问,也就是代理对象控制对原对象的引⽤。在某些情况下,⼀个对象不适合或者不能直接被引⽤访问,⽽代理对象可以在客⼾端和⽬标对象之间起到中介的作⽤。 二、为什么使用代理模式 模式作…

MySQL各种锁

目录 1. 从粒度上区分锁 1.1 全局锁(第一粒度) 1.2 表级锁(第二粒度) 1.3 行锁(第三最小粒度) 2 从模式上区分锁 2.1 什么是乐观锁 2.2 什么是悲观锁 2.3 意向共享锁和意向排他锁 2.4 临键锁和记录…

【Python】 深入理解Python中的UnicodeDecodeError及其解决方案

基本原理 在Python编程中,我们经常需要处理各种类型的数据,尤其是文本数据。文本数据在计算机中通常以字节的形式存在,而字节需要被解码成我们能够理解的字符。这个过程涉及到编码和解码的概念。 编码是将字符转换为字节的过程,…

23 vue3面试重难点复习:响应式原理、特点、8大生命钩子、data数据定义、组件、全家桶

vue作为用的最为广泛的当前热门框架,总结如下重难点核心知识: 1.vue特点是什么? 1.1优点 渐进式 vue本身只提供数据响应式,需要全局缓存用 vuex,需要路由用 vue-router 组件化 封装组件,利于复用 响应式数…

k8s——Pod进阶(资源限制和探针)

一、资源限制 1.1 资源限制的定义 当定义Pod时可以选择性地为每个容器设定所需要的资源数量。 最常见的可设定资源是CPU和内存大小,以及其他类型的资源。 当为Pod中的容器指定了request资源时,调度器就使用该信息来决定将Pod调度到哪个节点上。当还为容器…

汇凯金业:量化交易有风险吗

量化交易是一种通过复杂的数学模型和算法在金融市场中进行高频和自动化交易的方式。尽管量化交易在提高市场效率、减少人为错误等方面具有诸多优点,但它也同样存在着不少风险。以下列举了一些主要的风险因素: 1. 模型风险 模型缺陷:量化交易…

网络协议。

一、流程案例 接下来揭秘我要说的大事情,“双十一”。这和我们要讲的网络协议有什么关系呢? 在经济学领域,有个伦纳德里德(Leonard E. Read)创作的《铅笔的故事》。这个故事通过一个铅笔的诞生过程,来讲述…

数据安全之翼:天空卫士在汽车数据安全领域的卓越领航

近期,中国汽车网络安全与数据安全产业的积极倡导者谈思实验室发布首份《汽车网络与数据安全行业全景图》,天空卫士入选,并且位列榜首。 天空卫士在汽车数据安全领域有丰富的实践经验,曾为多家汽车行业用户提供数据安全产品与服务&…

LeetCode - 贪心(Greedy)算法集合(Python)[分配问题|区间问题]

欢迎关注我的CSDN:https://spike.blog.csdn.net/ 本文地址:https://spike.blog.csdn.net/article/details/139242199 贪心算法,是在每一步选择中,都采取当前状态下,最好或最优(即最有利)的选择&…

不同linux账户切换不同的cuda版本

原因 由于服务器中安装了两个版本的cuda(cuda10.1和cuda11.1),不同项目可能需要应用不同的cuda版本,但是自己又没有root权限或者只想在使用指定conda环境时改为用指定的cuda版本。总结起来有三种方法: 1、修改软链接指…

正点原子[第二期]Linux之ARM(MX6U)裸机篇学习笔记-24.1,2 SPI驱动实验-SPI协议介绍

前言: 本文是根据哔哩哔哩网站上“正点原子[第二期]Linux之ARM(MX6U)裸机篇”视频的学习笔记,在这里会记录下正点原子 I.MX6ULL 开发板的配套视频教程所作的实验和学习笔记内容。本文大量引用了正点原子教学视频和链接中的内容。…

Linux实验六:进程间通信(二)

目录 一、实验目的二、实验内容三、实验环境四、参考代码五、实验步骤步骤1. 编辑源代码test6.c步骤2. 编译源代码test6.c步骤3. 运行可执行程序test6步骤4. 进一步调试源代码test6.c 六、实验结果七、实验总结 一、实验目的 1、理解 POSIX 和 System V 提供的 IPC 相关概念&a…

安防监控视频平台LntonCVS视频监控汇聚平台遏制校园暴力保护校园学生安全应用方案

未成年人被誉为祖国的花朵,是我们国家的未来。然而,最近频繁曝出的未成年霸凌事件却引发了社会的广泛关注。这些事件手段残忍,事态恶劣,引发了全社会对如何保护未成年身心健康、规避霸凌事件发生的深刻思考。 为了更好地保障学生的…

从零开始:如何用Electron将chatgpt-plus.top 打包成EXE文件

文章目录 从零开始:如何用Electron将chatgpt-plus.top 打包成EXE文件准备工作:Node.js和npm国内镜像加速下载初始化你的Electron项目创建你的Electron应用运行你的Electron应用为你的应用设置图标打包成EXE文件结语 从零开始:如何用Electron将…

echarts学习:将echats实例代理为响应式对象可能带来的风险

1.起源 最近我在学习如何封装echarts组件,我所参考的其中一篇博客中提到了一个“图表无法显示的问题”。 根据其中的介绍,造成此种问题的原因是因为,使用ref接受了echarts实例,使得echarts实例被代理为了响应式对象,进…

[C#]使用C#部署yolov8的obb旋转框检测tensorrt模型

【测试通过环境】 win10 x64 vs2019 cuda11.7cudnn8.8.0 TensorRT-8.6.1.6 opencvsharp4.9.0 .NET Framework4.7.2 NVIDIA GeForce RTX 2070 Super 版本和上述环境版本不一样的需要重新编译TensorRtExtern.dll,TensorRtExtern源码地址:TensorRT-CShar…

3D视觉系统实现自动化上下料操作

在竞争激烈的汽车制造行业,提高生产效率、降低成本并保证产品质量是企业持续发展的关键。特别是在汽车制造过程中,各种零部件的上下料操作占据了大量的生产时间,因此如何实现这些操作的自动化、高效化成为了一个亟待解决的问题。 富唯智能3D视…

pom文件中,Maven导入依赖出现 Dependency not found

解决方案: 1、首先看一下自己的Maven是否配置好了 2、再检查一下镜像是否正确 3、如果上面都没有问题,看 dependencyManagement 标签 我这个出错,爆一大片红就是因为 这个标签 dependencyManagement 解决方法:在父工程中进行依…

在 Kubesphere 中开启新一代云原生数仓 Databend

上周六,由 KubeSphere 社区联合 Databend 社区以及纵目科技共同组织的云原生 Meetup 北京站在北京圆满落幕。本次 Meetup 活动邀请到了 SkyWalking PMC 成员、青云科技架构及可观测性团队负责人、江苏纵目科技 APM 研发总监、青云科技容器产品经理、数元灵科技 CTO …

JVM内存划分类加载的过程双亲委派模型的详解

JVM内存划分 JVM也就是java进程,这个进程一旦跑起来就会从操作系统这里申请一大块内存空间,JVM接下来就要进一步的对这个大的空间进行划分,划分成不同区域,从而每个区域都有不同的功能作用,一共分为如下几个区域 1.堆…