想要看更加舒服的排版、更加准时的推送
关注公众号“不太灵光的程序员”
每日八点有干货推送

有粉丝留言问什么时候可以写一个关于自动化任务的文章
准备上！~
感觉有用关注公众号 “不太灵光的程序员”
每天9:30分析一些Python小知识，爬坑的路与你同行

什么是自动化任务

自动化任务就是将一系列可以手动去实现的操作的集合，在特定的条件下自动去执行，比较常见的场景：

运行监控报警，监测要素的变化，触发阈值条件时执行一系列通知邮件、短信的发送，也可以是更复杂的触发
自动化田间灌溉控制，对土壤含水量、天气预报、植物生长周期分析需水量来调节灌溉水量
自动驾驶，根据行驶路线、实时路况信息进行分析后实施自动的驾驶
自动化测试，对目标系统有已知的预期，按照配置的流程去验证每一个功能的结果
定时爬虫，比如我们需要长期的爬取招聘信息，来进行数据分析，就需要定时的去爬取某段时间的增量信息
还有些统计类问题，比如我们需要对请求的IP进行判断是否是一个坏的IP(同一个IP下有一定比例的用户从事违规操作时，判断为非法IP)，我们可以在用户请求是实时去计算坏IP的列表，但这当用户量非常大的时候就不是一个很好的选择了，我们可以选择定期执行这个坏IP列表的计算，因为对于不同用户的坏IP列表是同一个

以上这些任务都可以认为是自动化任务，我们今天主要针对其中的定时任务进行讨论。

定时任务

定时任务的实现方式又有N多种，我就了解的几种实现举例说明：

Windows操作系统的任务计划程序
Linux操作系统的 Crontab 定时任务
Python的循环+sleep
Python的 Timer
Python的 schedule
Python的 APScheduler
Python的 Celery

我以两个场景来设计实例代码

每30秒打印一下当前时间
在2020-06-21 00:00:00 （具体时刻）打印一下当前时间

1. Windows任务计划程序

当前版本win10

右键“计算机”->选择“管理”;
然后就可以打开 “计算机管理”界面;
在界面的左侧有点击“系统工具”->任务计划程序 ;
可以看到右侧有“操作”“任务计划程序”;
有创建基本任务、创建任务、导入任务;
我以两个场景来设计实例代码
每30秒打印一下当前时间
在2020-06-21 00:00:00 （具体时刻）打印一下当前时间

任务1：由于任务计划程序最小执行周期是“5分钟”，是在高级设置中设置重复任务间隔，并选择持续时间“无限制” 无法实现30秒的打印
任务2：设置中选择一次，开始时间设置2020-06-21 00:00:00 （具体时刻）可以满足需求

2. Linux Crontab定时任务

Crontab定时任务，在每个任务周期中执行一次特定任务，固无法控制执行次数
时间格式如下：

*    *    *    *    *  program
-    -    -    -    -    -
|    |    |    |    |    |----- 要执行的程序
|    |    |    |    +----- 星期中星期几 (0 - 7) (星期天 为0) 默认*
|    |    |    +---------- 月份 (1 - 12) 默认*
|    |    +--------------- 一个月中的第几天 (1 - 31) 默认*
|    +-------------------- 小时 (0 - 23) 默认*
+------------------------- 分钟 (0 - 59) 默认*

我以两个场景来设计实例代码

每30打印一下当前时间;
在2020-06-21 00:00:00 （具体时刻）打印一下当前时间 ;

任务1：* * * * * echo $(date +%F%n%T)含义每分钟打印当前时间，最小执行周期每分钟 无法实现30s的打印
任务2：* * 21 6 * echo $(date +%F%n%T)含义每年的6月21日打印当前时间 无法实现只执行一次的打印

3. Python 循环+sleep

我以两个场景来设计实例代码

每30打印一下当前时间;
在2020-06-21 00:00:00 （具体时刻）打印一下当前时间 ;

任务1： 可以实现30s的打印
借助循环和等待时间来实现任务执行，但是在一个线程组阻塞时不能进行其他操作，可以使用协程优化

import time
import datetime# 每30秒执行
while 1:time.sleep(30)print(datetime.datetime.now())

任务2： 可以实现特定时间的打印
每次循环判断当前时刻和目标是否一致，需要注意执行一次就需要跳出点前时刻的判断，不然在同一时刻会重复打印很多次

import time
import datetim# 特定时间执行
while 1:now = datetime.datetime.now()if now.year == 2020 and now.month == 6 and now.day == 21:print(now)breaktime.sleep(1)

4. 线程Timer

看下源码的描述

class Timer(Thread):"""Call a function after a specified number of seconds:t = Timer(30.0, f, args=None, kwargs=None)t.start()t.cancel()     # stop the timer's action if it's still waiting"""

在指定的秒数后调用函数
因为创建的定时是异步执行，所以不存在等待顺序执行问题。

创建定时器 Timer(interval, function, args=None, kwargs=None);
取消定时器cancel();
使用线程方式执行start();
等待线程执行结束join(self, timeout=None);
定时器只能执行一次，如果需要重复执行，需要重新添加任务;

我以两个场景来设计实例代码

每30打印一下当前时间
在2020-06-21 00:00:00 （具体时刻）打印一下当前时间

任务1： 可以实现30s的打印
虽然是可以实现没30秒一次的打印，但是定时器只能执行一次，如果需要重复添加任务，创建线程又比较浪费系统资源

from threading import Timer
import datetime
import timedef run_time(seconds, flag=False):t = Timer(seconds, run_time, [seconds, True])t.start()if flag:print(datetime.datetime.now())time.sleep(3)t.join()run_time(30)

任务2： 实现特定时间的打印
需要计算当前时间到目标时间间的秒数来实现

from threading import Timer
import datetime
import timedef run_time(seconds, flag=False):if seconds > 0:t = Timer(seconds, run_time, [seconds, True])t.start()if flag:print(datetime.datetime.now())time.sleep(3)t.join()def get_seconds(run_date):start = datetime.datetime.now()end = datetime.datetime.strptime(run_date, "%Y-%m-%d %H:%M:%S")if end > start:return (end-start).secondselse:return -1seconds = get_seconds("2020-06-21 00:00:00")
run_time(seconds)

5. schedule模块

首先看下源码的描述

Python job scheduling for humans.github.com/dbader/scheduleAn in-process scheduler for periodic jobs that uses the builder pattern
for configuration. Schedule lets you run Python functions (or any other
callable) periodically at pre-determined intervals using a simple,
human-friendly syntax.Inspired by Addam Wiggins' article "Rethinking Cron" [1] and the
"clockwork" Ruby module [2][3].Features:- A simple to use API for scheduling jobs.- Very lightweight and no external dependencies.- Excellent test coverage.- Tested on Python 2.7, 3.5 and 3.6Usage:>>> import schedule>>> import time>>> def job(message='stuff'):>>>     print("I'm working on:", message)>>> schedule.every(10).minutes.do(job)>>> schedule.every(5).to(10).days.do(job)>>> schedule.every().hour.do(job, message='things')>>> schedule.every().day.at("10:30").do(job)>>> while True:>>>     schedule.run_pending()>>>     time.sleep(1)

任务1： 可以实现30s的打印
可以实现没30秒一次的打印，还可以同时创建多个定时任务同时去执行，由于执行过程是顺序执行，pn休眠2秒，循环任务查询休眠1秒，会存在实际间隔时间并不是设定的30秒，当工作任务回非常耗时就会影响其他任务的触发时间

import datetime
import schedule
import timedef pn():print(datetime.datetime.now())time.sleep(2)schedule.clear()
schedule.every(30).seconds.do(pn)while 1:schedule.run_pending()time.sleep(1)

任务2： 实现特定时间的打印
需要计算当前时间到目标时间间的秒数来实现

import datetime
import schedule
import timedef pn():print(datetime.datetime.now())time.sleep(2)def get_seconds(run_date):start = datetime.datetime.now()end = datetime.datetime.strptime(run_date, "%Y-%m-%d %H:%M:%S")if end > start:return (end - start).secondselse:return -1schedule.clear()
seconds = get_seconds("2020-06-21 00:00:00")
if seconds > 0:schedule.every(seconds).seconds.do(pn)while 1:schedule.run_pending()time.sleep(1)

6. APScheduler定时任务框架

APScheduler是Python的一个定时任务框架，用于执行周期或者定时任务，
可以基于日期、时间间隔，及类似于Linux上的定时任务crontab类型的定时任务；
该该框架不仅可以添加、删除定时任务，还可以将任务存储到数据库中，实现任务的持久化，使用起来非常方便。

triggers（触发器）：触发器包含调度逻辑，每一个作业有它自己的触发器;
job stores（作业存储）:用来存储被调度的作业，默认的作业存储器是简单地把作业任务保存在内存中，支持存储到MongoDB，Redis数据库中;
executors（执行器）：执行器用来执行定时任务，只是将需要执行的任务放在新的线程或者线程池中运行;
schedulers（调度器）：调度器是将其它部分联系在一起,对使用者提供接口，进行任务添加，设置，删除。

add_job 创建任务

创建任务支持三种类型触发器date、interval和cron：

date触发器

run_date 具体的日期执行，时间参数run_date，可以是dete/time格式的字符串，当不写时分秒是默认是 00:00:00

interval触发器

weeks 参数weeks 每n周后执行，int类型
days 参数days每n天后执行，int类型
hours: 参数hours每n小时后执行，int类型
minutes:参数minutes每n分钟后执行，int类型
seconds: 参数seconds每n秒后执行，int类型
datetime|str 参数start_date，可以是dete/time格式的字符串，控制执行循环的时间范围
datetime|str 参数 end_date，可以是dete/time格式的字符串，控制执行循环的时间范围

cron触发器

使用方式与 Crontab定时任务类似

year int类型取值范围1970~9999，默认*
month int类型取值范围1~12，默认1
dayint类型取值范围1~31，默认1
week 一年中的第几周，int类型取值范围1~53，默认*
day_of_week一周中的第几天，int类型取值范围0~6，默认*
hour int类型取值范围0~23，默认0
minute int类型取值范围0~59，默认0
second int类型取值范围0~59，默认0

任务1： 可以实现30s的打印
使用interval类型触发器添加定时任务
可以实现没30秒一次的打印，还可以同时创建多个定时任务同时去执行，由于执行过程是顺序执行，pn休眠3秒，会存在实际间隔时间并不是设定的30秒，当工作任务回非常耗时就会影响其他任务的触发时间

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
import timedef pn():print(datetime.datetime.now())time.sleep(3)def run_time(seconds):scheduler = BlockingScheduler()scheduler.add_job(pn, 'interval', seconds=seconds, id=str(time.time()))scheduler.start()run_time(30)

任务2： 可以实现特定时间的打印
使用date类型或cron类型触发器添加定时任务

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
import timedef pn():print(datetime.datetime.now())time.sleep(3)def run_time():scheduler = BlockingScheduler()scheduler.add_job(pn, 'cron', year=2020, month=6, day=21, id=str(time.time()))scheduler.add_job(pn, 'date', run_date="2020-06-21", id=str(time.time()))scheduler.start()run_time()

7. Celery分布式任务调度的框架

Celery是实时处理的异步任务调度的框架，需要借助中间件完成调度任务，也可以很好的用来做定时任务服务。

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统，由三部分组成：

消息中间件（message broker）：Celery本身不提供消息服务，依赖于中间件RabbitMQ, Redis等
任务执行单元（worker）: 是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中
任务执行结果存储（task result store）: 用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, Redis等

以上面代码使用的 python3.7不同
当前版本 python2.7 Celery 3.2

目录结构

celery_app_init_.py

# -*- coding: utf-8 -*-
# 拒绝隐式引入，如果celery.py和celery模块名字一样，避免冲突，需要加上这条语句
# 该代码中，名字是不一样的，最好也要不一样
from __future__ import absolute_import
from celery import Celeryapp = Celery('tasks')
app.config_from_object('celery_app.celeryconfig')

celery_app\celeryconfig.py

# -*- coding: utf-8 -*-
from __future__ import absolute_import
from celery.schedules import crontab
from datetime import timedelta# 使用redis存储任务队列
BROKER_URL = 'redis://127.0.0.1:6379/1'
# 使用redis存储结果
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/2'# 时区设置
CELERY_TIMEZONE = 'Asia/Shanghai'# 导入任务所在文件
CELERY_IMPORTS = ['celery_app.celery_task.tasks',
]# 需要执行任务的配置
CELERYBEAT_SCHEDULE = {'np-seconds-10': {# 具体需要执行的函数# 该函数必须要使用@app.task装饰'task': 'celery_app.celery_task.tasks.pn',# 定时时间# 每30秒分钟执行一次'schedule': timedelta(seconds=30),'args': ()},'np-month-day': {'task': 'celery_app.celery_task.tasks.pn',# 每年6月21日执行一次'schedule': crontab(day_of_month=21, month_of_year=6),'args': ()},
}

celery_app\celery_task\tasks.py

# -*- coding: utf-8 -*-
from .. import app
import datetime
import time@app.task
def pn():print(datetime.datetime.now())time.sleep(3)

我以两个场景来设计实例代码

每30打印一下当前时间
在2020-06-21 00:00:00 （具体时刻）打印一下当前时间

任务1： 可以实现30s的打印

    'np-seconds-10': {# 具体需要执行的函数# 该函数必须要使用@app.task装饰'task': 'celery_app.celery_task.tasks.pn',# 定时时间# 每30秒分钟执行一次'schedule': timedelta(seconds=30),'args': ()},

任务2： 不可以实现指定时间的打印
一般指定时间也没有意思，每年的某个月日的任务比较多
使用方式与 Crontab定时任务类似

    'np-month-day': {'task': 'celery_app.celery_task.tasks.pn',# 每年6月21日执行一次'schedule': crontab(day_of_month=21, month_of_year=6),'args': ()},

启动定时任务

由于是分布式服务，我们需要启动中间件，触发器和处理模块

start redis

celery_app 统计目录下执行

发布任务

celery -A celery_app beat

执行任务

celery worker -A celery_app -l info

总结

简单总结上面七种定时定点任务实现：

Windows任务计划程序合适重启服务，定时执行些bat脚本，清理日志等操作；
Linux Crontab定时任务适合执行shell脚本或者一系列Linux命令；
循环+sleep方式适合简单测试；
Timer可以实现定时任务，但是对定点任务来说，需要检查当前时间点；
schedule可以定点定时执行，但是需要在循环中检测任务，而且存在阻塞；
APScheduler框架更加强大，可以直接在里面添加定点与定时任务；
Celery框架稳定，可动态添加任务，与Django、Flask等框架可以完美结合使用，对于大任务量的服务支持比较好。

【Python自动化任务】让运维更简单的7种定时任务实现方式，总有一种适合你的场景相关推荐

运维更简单、更智能，让运维人不再 “拼命”
简介: 云原生智能运维解决方案,利用大数据为企业日常运维服务,通过可观测数据,融合智能告警与响应中枢,结合机器学习的方法进一步解决自动化运维所未解决的问题,让运维更简单.更智能. 在90%的科幻片中 ...
微服务平台（Micro Service Platform : MSP）旨在提供一个集开发、测试、运维于一体的开发者专属平台，让开发者能快速构建或使用微服务，让开发更简单，让运维更高效。...
微服务平台(Micro Service Platform : MSP)旨在提供一个集开发.测试.运维于一体的开发者专属平台,让开发者能快速构建或使用微服务,让开发更简单,让运维更高效. MSP采用业界 ...
python比java简单好学-python和java哪个学起来更简单
在近几年Python的呼声越来越高,很多刚开始起步想要学习编程的朋友都会犹豫要不要选择学习Python,毕竟作为人工智能时代的首选语言这个诱惑还是很大的.在选择上最纠结的就是Python和Java选择 ...
python和java选择哪个-python和java哪个学起来更简单
在近几年Python的呼声越来越高,很多刚开始起步想要学习编程的朋友都会犹豫要不要选择学习Python,毕竟作为人工智能时代的首选语言这个诱惑还是很大的.在选择上最纠结的就是Python和Java选择 ...
让运维更高效：关于ECS系统事件
摘要: 阿里云会针对ECS实例发布系统事件,当您收到阿里云计划维护的通知时,可以利用ECS系统计划事件了解与实例相关的计划维护操作,并根据您的业务特性选择合适的时间安排运维操作进行故障转移,减少对系统 ...
容器化技术在运维管理中的应用：自动化、可视化运维
作者:禅与计算机程序设计艺术容器化技术在运维管理中的应用:自动化.可视化运维引言随着云计算和互联网的发展,容器化技术作为一种轻量级.高效的编程方式,得到了越来越广泛的应用.在容器化技术的应用场景 ...
puppet、Ansible、SaltStack 自动化运维工具简单对比
puppet: 有产品线已经在用,优点是历史悠久,比较成熟,在可远程可本地,功能强劲,不过这厮批量执行功能没得,为了批量执行个命令写个配置文件,好像有点大刀砍蚊子腿的感觉了,而且有客户端在,和授权系统 ...
python与linux系统运维_《Python Linux系统管理与自动化运维》碎碎念
在此记录学习<Python Linux系统管理与自动化运维>的过程感觉碎碎念全放在一篇文章里编辑时太乱,于2020年12月16日重新分开整理. 2020年12月13日,看了一下需要学习的 ...
运维自动化之---ansilbe运维自动化和ansible架构介绍（1）
运维自动化的发展历程 1.自动化运维应用场景 1.1云计算运维工程师核心职能运维相关的工具 Podman是用来替代docker的工具 1.2 运维职业的发展路线目标::一切皆自动化 1.3 企业实 ...

【Python自动化任务】让运维更简单的7种定时任务实现方式，总有一种适合你的场景

什么是自动化任务

定时任务

1. Windows任务计划程序

2. Linux Crontab定时任务

3. Python 循环+sleep

4. 线程Timer

5. schedule模块

6. APScheduler定时任务框架

add_job 创建任务

date触发器

interval触发器

cron触发器

7. Celery分布式任务调度的框架

目录结构

celery_app_init_.py

celery_app\celeryconfig.py

celery_app\celery_task\tasks.py

启动定时任务

start redis

发布任务

执行任务

总结

推荐阅读：

【Python自动化任务】让运维更简单的7种定时任务实现方式，总有一种适合你的场景相关推荐

最新文章

热门文章