Python使用定时调度任务的方式

目录
  • 1、简单循环 Simple loops
  • 2、简单循环但是使用了线程Simple loops but threaded
  • 3、定时调度库 Schedule Library
  • 4、Python Crontab
  • 5、RQ 调度器 RQ Scheduler
  • 6、总结 Conclusion

摘要:

今天构建的大多数应用程序都需要某种方式的调度机制。轮询 API 或数据库、不断检查系统健康状况、将日志存档等是常见的例子。 KubernetesApache Mesos等使用自动伸缩扩容技术(Auto-scaling)的软件需要检查部署的应用程序的状态,为此它们使用定期运行的存活探针(Liveness Probe)。调度任务需要与业务逻辑解耦,因此我们要使用解耦的执行队列,例如Redis队列。

Python 有几种方法可以定时调度一个任务,这就是我们将在本文中学习的内容。我将使用以下方式讨论调度任务:

  • 简单循环 (Simple Loops
  • 简单循环但是使用了线程 (Simple Loops but Threaded
  • 调度库 (Schedule Library
  • Python Crontab
  • RQ 调度器作为解耦队列 (RQ Scheduler as decoupled queues

1、简单循环 Simple loops

使用简单循环来实现调度任务这是毫不费力的。使用无限运行的 while 循环定期调用函数可用于调度作业,但这不是最好的方法,不过它是很有效的。可以使用内置time模块的slleep()来延迟执行。不过这并不是大多数作业的调度方式,因为,它看起来很难看,而且与其他方法相比,它的可读性较差。

import time
​
def task():
    print("Job Completed!")
​
 while 1:
    task()
    time.sleep(10)

当涉及到每天早上 9:00 或每周三晚上 7:45 等这些日程安排时,事情就变得比较棘手了。

import datetime
​
def task():
    print("Job Completed!")
​
 while 1:
    now = datetime.datetime.now()
    # schedule at every wednesday,7:45 pm
    if now.weekday == 3 and now.strftime("%H:%m") == "19:45":
        task()
    # sleep for 6 days
    time.sleep(6 * 24 * 60 * 60)

这是我的第一时间想到的解决办法,不用谢!这种方法的一个问题是这里的逻辑是阻塞的,即一旦在 python 项目中发现这段代码,它就会卡在 while 1 循环中,从而阻塞其他代码的执行。

2、简单循环但是使用了线程Simple loops but threaded

线程是计算机科学中的一个概念。具有自己指令的小程序由进程执行并独立管理,这就可以解决我们第一种方法的阻塞情况,让我们看看怎么样。

import time
import threading
​
def task():
    print("Job Completed!")
​
def schedule():
    while 1:
        task()
        time.sleep(10)
​
# makes our logic non blocking
thread = threading.Thread(target=schedule)
thread.start()

线程启动后,其底层逻辑无法被主线程修改,因此我们可能需要添加资源,程序通过这些资源可以检查特定场景并根据它们执行逻辑。

3、定时调度库 Schedule Library

早些时候,我说使用 while 循环进行调度看起来很丑陋,调度库可以解决这个问题。

import schedule
import time
​
def task():
    print("Job Executing!")
​
# for every n minutes
schedule.every(10).minutes.do(task)
​
# every hour
schedule.every().hour.do(task)
​
# every daya at specific time
schedule.every().day.at("10:30").do(task)
​
# schedule by name of day
schedule.every().monday.do(task)
​
# name of day with time
schedule.every().wednesday.at("13:15").do(task)
​
while True:
    schedule.run_pending()
    time.sleep(1)

正如您所见,通过这样我们可以毫不费力地创建多个调度计划。我特别喜欢创建作业的方式和方法链(Method Chaining),另一方面,这个片段有一个 while 循环,这意味着代码被阻塞,不过我相信你已经知道什么可以帮助我们解决这个问题。

4、Python Crontab

Liunx 中的 crontab 实用程序是一种易于使用且被广泛接受的调度解决方案。Python python-crontab提供了一个 API 来使用 Python 中的 CLI 工具。在crontab中,一个定时调度使用 unix-cron字符串格式( *)来描述,它是一组五个值的一条线,这表明当作业应该被执行时,python-crontab 将在文件中写入 crontab 的计划转换为写入编程方法。

from crontab import CronTab
​
cron = CronTab(user='root')
​
job = cron.new(command='my_script.sh')
​
job.hour.every(1)
cron.write()

python-crontab 不会自动保存计划,需要执行 write() 方法来保存计划。还有更多功能,我强烈建议您查看他们的文档。

5、RQ 调度器 RQ Scheduler

有些任务不能立即执行,因此我们需要根据 LIFO FIFO 等队列系统创建任务队列并弹出任务。python-rq允许我们做到这一点,使用 Redis 作为代理来排队作业。新作业的条目存储为带有信息的哈希映射,例如created_at, enqueued_at, origin, data, description.

排队任务由名为 worker 的程序执行。workers Redis 缓存中也有一个条目,负责将任务出列以及更新 Redis 中的任务状态。任务可以在需要时排队,但要安排它们,我们需要rq-scheduler

from rq_scheduler import Scheduler
​
queue = Queue('circle', connection=Redis())
scheduler = Scheduler(queue=queue)
​
scheduler.schedule(
    scheduled_time=datetime.utcnow(), # Time for first execution, in UTC timezone
    func=func,                     # Function to be queued
    args=[arg1, arg2],             # Arguments passed into function when executed
    kwargs={'foo': 'bar'},         # Keyword arguments passed into function when executed
    interval=60,                   # Time before the function is called again, in seconds
    repeat=None,                     # Repeat this number of times (None means repeat forever)
    meta={'foo': 'bar'}            # Arbitrary pickleable data on the job itself
)

RQ worker(RQ 工作器)必须在终端中单独启动或通过 python-rq 工作器启动。一旦任务被触发,就可以在工作终端中看到,在成功和失败场景中都可以使用单独的函数回调。

6、总结 Conclusion

还有一些用于调度的库,但在这里,我已经讨论了最常见的库。值得一提的是Celerycelery 的另一个优点是用户可以在多个代理之间进行选择。我很感激你读到最后。也可以看看我的其他文章。干杯!

到此这篇关于Python使用定时调度任务的方式的文章就介绍到这了,更多相关Python使用定时调度任务方式内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • python实现定时发送邮件到指定邮箱

    本文实例为大家分享了python实现定时发送邮件到指定邮箱的具体代码,供大家参考,具体内容如下 整个链路:传感器采集端采集数据,边缘端上传数据库,从数据库拿到数据. 产品端有个自动出报告的需求,并且希望自动出的报告可以定时通过邮件发到指定邮箱中. 自动发邮件中也有较多坑,本文特此记录自动发邮件中的坑. 使用:Python3.7,win10系统,163邮箱作为发邮件方,QQ邮箱作为收邮件方. **第一步:**申请两个账号分别作为发送和接收的账号,本文用的是一个163邮箱,一个是qq邮箱. 第一个坑

  • python脚本定时发送邮件

    本文实例为大家分享了python定时发送邮件的具体代码,供大家参考,具体内容如下 全部代码如下: import time from datetime import datetime from email.header import Header from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import parseaddr, formatad

  • python编写网页爬虫脚本并实现APScheduler调度

    前段时间自学了python,作为新手就想着自己写个东西能练习一下,了解到python编写爬虫脚本非常方便,且最近又学习了MongoDB相关的知识,万事具备只欠东风. 程序的需求是这样的,爬虫爬的页面是京东的电子书网站页面,每天会更新一些免费的电子书,爬虫会把每天更新的免费的书名以第一时间通过邮件发给我,通知我去下载. 一.编写思路: 1.爬虫脚本获取当日免费书籍信息 2.把获取到的书籍信息与数据库中的已有信息作比较,如果书籍存在不做任何操作,书籍不存在,执行插入数据库的操作,把数据的信息存入Mo

  • Python实现FTP文件定时自动下载的步骤

    之前遇到技术问题总能在技术博客上得到启发,十分感谢各位的无私分享.而自己却很少发文,固然是水平有限,但也限制了知识积累和总结.今后多总结分享,回馈博客的同时也希望大家多多批评. 一.需求: 某数据公司每日15:00~17:00之间,在其FTP发布当日数据供下载,我方需及时下载当日数据至指定本地目录. 二.分析: 1.需实现FTP登陆.查询.下载功能: 解答:使用内置的ftplib模块中FTP类: 2.需判断文件是否下载: 解答:使用os模块中path.exists方法: 3.需判断在指定时间段内

  • scrapy处理python爬虫调度详解

    学习了简单的知识点,就会想要向有难度的问题挑战,这里必须要夸一夸小伙伴们.不过我们今天不需要做什么程序的测试,只用简单的两个代码对比,小伙伴们就能在其中体会两者的不同和难易程度.scrapy能否适合处理python爬虫调度的问题,小编直接说出答案小伙伴们也不能马上信服,下面就让我们在示例中找寻答案吧. 总的来说,需要使用代码来爬一些数据的大概分为两类人: 非程序员,需要爬一些数据来做毕业设计.市场调研等等,他们可能连 Python 都不是很熟: 程序员,需要设计大规模.分布式.高稳定性的爬虫系统

  • 自动在Windows中运行Python脚本并定时触发功能实现

    目录 环境介绍 运行Python脚本:.bat文件 举例: 定时在Windows中触发.bat文件(英文版系统) 环境介绍 系统环境:Windows 10 Python版本:Python 3.5 必备包:无 运行Python脚本:.bat文件 在Windows中,.bat文件是批处理文件,是与Linux中.sh(shell)文件很像的东西. 如果,我们想在Windows中运行一个Python脚本,我们可以通过CMD,首先进入python文件所在的目录,之后运行. 但是这样很麻烦,每次都要打开CM

  • 如何给windows设置定时任务并运行python脚本

    目录 1. 创建定时任务 1.1 计划任务 2.2 基本任务 2.3 命名 2.4 执行时间 2.5执行时间 2.6 启动程序 2.7 选择启动程序 2.8完成 3. 结尾 4. 完工 自己用python写了一个签到脚本,经过测试已经可以成功打卡,于是研究了一下windows定时运行程序 1. 创建定时任务 1.1 计划任务 打开"控制面板 "–>"系统和安全"–>"管理工具"–>"计划任务",如图 2.2

  • 简单的Python调度器Schedule详解

    最近在做项目的时候经常会用到定时任务,由于我的项目是使用Java来开发,用的是SpringBoot框架,因此要实现这个定时任务其实并不难. 后来我在想如果我要在Python中实现,我要怎么做呢? 一开始我首先想到的是Timer Timer 这个是一个扩展自threading模块来实现的定时任务.它其实是一个线程. # 首先定义一个需要定时执行的方法 >>> def hello(): print("hello!") # 导入threading,并创建Timer,设置1秒

  • Python爬虫定时计划任务的几种常见方法(推荐)

    记得以前的Windows任务定时是可以正常使用的,今天试了下,发现不能正常使用了,任务计划总是挂起.接下来记录下Python爬虫定时任务的几种解决方法. 1.方法一.while True 首先最容易的是while true死循环挂起,不废话,直接上代码: import os import time import sys from datetime import datetime, timedelta def One_Plan(): # 设置启动周期 Second_update_time = 24

  • python爬虫调度器用法及实例代码

    我们一般使用爬虫看到的都是最后的数据结果,对于整个的获取过程没有过多了解过.对于初学python的小伙伴们来说,不光是代码的练习,还是原理的分析都是必不可少的. 小编把整个爬取的过程分为了几个部分,从一开始的下载,到数据的去重解析,再到整个爬虫循环的结束,以图片和代码的双重形式展现给大家,希望能够对爬虫调度器有一个深刻的理解. 我们可以编写几个元件,每个元件完成一项功能,下图中的蓝底白字就是对这一流程的抽象: UrlManager:将存储和获取url以及url去重的几个步骤在url管理器中完成(

  • python 实现定时任务的四种方式

    用Python实现定时任务 有些时候我们需要每隔一段时间就要执行一段程序,或者是往复循环执行某一个任务.比如博主在上篇文章讲的爬虫一样,在实现对某个目标进行在线爬取的话,也需要用到实时任务. 用Python实现定时任务的四种方法 while True: + sleep() threading.Timer定时器 调度模块schedule 任务框架APScheduler 定时要完成的Task(简单定义下) import datetime def Task(): now = datetime.date

  • python中用Scrapy实现定时爬虫的实例讲解

    一般网站发布信息会在具体实现范围内发布,我们在进行网络爬虫的过程中,可以通过设置定时爬虫,定时的爬取网站的内容.使用python爬虫框架Scrapy框架可以实现定时爬虫,而且可以根据我们的时间需求,方便的修改定时的时间. 1.Scrapy介绍 Scrapy是python的爬虫框架,用于抓取web站点并从页面中提取结构化的数据.任何人都可以根据需求方便的修改.Scrapy用途广泛,可以用于数据挖掘.监测和自动化测试. 2.使用Scrapy框架定时爬取 import time from scrapy

  • python获取指定时间段内特定规律的日期列表

    前言:写小程序时,遇到个需要自定义回归时间的场景,下面是实现的效果图. 于我需求而言,我只需要用到fixed_interval.weekly.per_month三个方法返回的回归间隔列表就行,最后的这个make_date_from_list方法,是为了能在这里展示效果 TIP:没有做异常处理,以及时间格式化之类的.这里只是抛砖引玉给个原型砖头出来. 实现代码: class GetStrategyDedail: today = datetime.date.today() def __init__(

  • 实现Windows下设置定时任务来运行python脚本

    1. 自己用python写了一个签到脚本 经过测试已经可以成功打卡,于是研究了一下windows定时运行程序 2. 创建定时任务 2.1 打开"控制面板 "–>"系统和安全"–>"管理工具"–>"计划任务" 如图 2.2 打开"计划任务",如图,点击"创建基本任务" 2.3 给定时任务命名,点击下一步 2.4 选择脚本执行的时间,以每天执行为例,如图,点击下一步 2.5

  • python实现定时发送邮件

    本文实例为大家分享了python实现定时发送邮件的具体代码,供大家参考,具体内容如下 一.发送邮件 import smtplib from email.mime.text import MIMEText from email.header import Header #引入smtplib.MIMETex和Header mailhost='smtp.qq.com' #把qq邮箱的服务器地址赋值到变量mailhost上,地址应为字符串格式 qqmail = smtplib.SMTP() #实例化一个

随机推荐