網站首頁 編程語言 正文
前言
在日常工作中，常常會用到需要周期性執行的任務，一種方式是采用 Linux 系統自帶的 crond 結合命令行實現。另外一種方式是直接使用Python。????????
當每隔一段時間就要執行一段程序，或者往復循環執行某一個任務，這就需要使用定時任務來執行程序。比如在實現對某個目標進行爬蟲的話，需要用到實時任務。
python中常用的定時任務主要有以下8種方法：
- while True：+sleep()
- threading.Timer定時器
- Timeloop庫執行定時任務
- 調度模塊sched
- 調度模塊schedule
- 任務框架APScheduler
- 分布式消息系統celery執行定時任務
- 使用windows自帶的定時任務
接下來分別用上述8中方式來完成下面定義的Task()任務，示例代碼如下：
from datetime import datetime
def task():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts)1、利用while True：+sleep()實現定時任務
最簡單的方式應該就是使用time模塊來實現定時任務，在循環里面放入要執行的任務，然后sleep一段時間再執行。實現令當前執行的線程暫停 n秒后再繼續執行。所謂暫停，即令當前線程進入阻塞狀態，當達到 sleep() 函數規定的時間后，再由阻塞狀態轉為就緒狀態，等待 CPU 調度。
示例代碼：
from datetime import datetime
import time
def task():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts)
def func():
while True:
task()
time.sleep(3)
func()運行結果：
優缺點：只能實現同步任務，無法執行異步任務。執行起來雖然是比較簡單，但不容易控制，而且sleep是個阻塞函數。只能設定間隔，不能指定具體的時間點。
2、利用threading.Timer()定時器實現定時任務
timer最基本理解就是定時器，可以啟動多個定時任務，這些定時器任務是異步執行，所以不存在等待順序執行問題。
| Timer方法 | 說明 |
|---|---|
| Timer(interval, function, args=None, kwargs=None) | 創建定時器 |
| cancel() | 取消定時器 |
| start() | 使用線程方式執行 |
| join(self, timeout=None) | 等待線程執行結束 |
示例代碼：
from datetime import datetime
from threading import Timer
def task():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts)
def func():
task()
t = Timer(3, func)
t.start()
func()運行結果：
優缺點：可以實現異步任務，是非阻塞的，但當運行次數過多時，會出現報錯：Pyinstaller maximum recursion depth exceeded Error Resolution 達到最大遞歸深度，然后想到的是修改最大遞歸深度，
sys.setrecursionlimit(100000000)但是運行到達到最大CPU時，python會直接銷毀程序。
3、使用Timeloop庫執行定時任務
Timeloop是一個庫，可用于運行多周期任務。這是一個簡單的庫，使用decorator模式在線程中運行標記函數。
示例代碼：
from datetime import datetime, timedelta
from timeloop import Timeloop
tl = Timeloop()
def task():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts + '333!')
def task2():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts + "555555!")
@tl.job(interval=timedelta(seconds=2))
def sample_job_every_2s():
task()
@tl.job(interval=timedelta(seconds=5))
def sample_job_every_5s():
task2()關于更多timeloop用法，詳見博文：???python中定時任務timeloop庫用法詳解
4、利用調度模塊sched實現定時任務
sched是一種調度（延時處理機制）。sched模塊實現了一個通用事件調度器，在調度器類使用一個延遲函數等待特定的時間，執行任務。同時支持多線程應用程序，在每個任務執行后會立刻調用延時函數，以確保其他線程也能執行。
scheduler對象主要方法：
- enter(delay, priority, action, argument)，安排一個事件來延遲delay個時間單位。
- cancel(event)：從隊列中刪除事件。如果事件不是當前隊列中的事件，則該方法將跑出一個ValueError。
- run()：運行所有預定的事件。這個函數將等待(使用傳遞給構造函數的delayfunc()函數)，然后執行事件，直到不再有預定的事件。
示例代碼：
import sched
import time
from datetime import datetime
# 初始化sched模塊的scheduler類
# 第一個參數是一個可以返回時間戳的函數，第二個參數可以在定時未到達之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)
def task(inc):
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts)
schedule.enter(inc, 0, task, (inc,))
def func(inc=3):
# enter四個參數分別為：
# 間隔事件、優先級（用于同時間到達的兩個事件同時執行時定序）、被調用觸發的函數、給該觸發函數的參數（tuple形式）
schedule.enter(0, 0, task, (inc,))
schedule.run()
func()運行結果：
關于更多sched用法，詳見博文：https://www.jb51.net/article/272340.htm
5、利用調度模塊schedule實現定時任務
schedule是一個第三方輕量級的任務調度模塊，可以按照秒，分，小時，日期或者自定義事件執行時間。
如果想執行多個任務，也可以添加多個task。
示例代碼：
import schedule
from datetime import datetime
def task():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts)
def task2():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts + '666!')
def func():
# 清空任務
schedule.clear()
# 創建一個按3秒間隔執行任務
schedule.every(3).seconds.do(task)
# 創建一個按2秒間隔執行任務
schedule.every(2).seconds.do(task2)
while True:
schedule.run_pending()
func()運行結果：
優缺點：需要和while Ture配合使用，而且占用的CPU也比其他幾種多的多，占用內存也是較大。
關于更多schedule用法，詳見博文：?https://www.jb51.net/article/272345.htm
6、利用任務框架ASPcheduler實現定時任務
APScheduler是Python的一個定時任務框架，用于執行周期或者定時任務，該框架不僅可以添加、刪除定時任務，還可以將任務存儲到數據庫中，實現任務的持久化，使用起來非常方便。
示例代碼：
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
def task():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts)
def task2():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts + '666!')
def func():
# 創建調度器BlockingScheduler()
scheduler = BlockingScheduler()
scheduler.add_job(task, 'interval', seconds=3, id='test_job1')
# 添加任務，時間間隔為5秒
scheduler.add_job(task2, 'interval', seconds=5, id='test_job2')
scheduler.start()
func()運行結果：
關于更多apschedule用法，詳見博文：python中定時任務apscheduler庫用法詳解
7、使用分布式消息系統celery執行定時任務
Celery是一個簡單，靈活，可靠的分布式系統，用于處理大量消息，同時為操作提供維護此類系統所需的工具, 也可用于任務調度。Celery 的配置比較麻煩，如果你只是需要一個輕量級的調度工具，Celery 不會是一個好選擇。
Celery 是一個強大的分布式任務隊列，它可以讓任務的執行完全脫離主程序，甚至可以被分配到其他主機上運行。我們通常使用它來實現異步任務（async task）和定時任務（crontab）。 異步任務比如是發送郵件、或者文件上傳, 圖像處理等等一些比較耗時的操作 ，定時任務是需要在特定時間執行的任務。
注意：celery本身并不具備任務的存儲功能，在調度任務的時候肯定是要把任務存起來的，因此在使用celery的時候還需要搭配一些具備存儲、訪問功能的工具，比如：消息隊列、Redis緩存、數據庫等。官方推薦的是消息隊列RabbitMQ，有些時候使用Redis也是不錯的選擇。
8、使用windows自帶的定時任務
略。這兒不做細述！
總結
原文鏈接：https://blog.csdn.net/weixin_44799217/article/details/127352531
相關推薦
- 2022-11-03 C++命名空間使用詳細介紹_C 語言
- 2022-06-15 C++詳細講解繼承與虛繼承實現_C 語言
- 2023-08-01 el-table 拖拽列寬出現空白列問題
- 2022-09-25 linux系統下oracle數據庫的導入導出
- 2023-08-13 fastadmin框架怎么重定向至后臺模塊
- 2022-05-03 python中的Pytorch建模流程匯總_python
- 2022-09-17 ASP.NET?Core項目中集成TypeScript_實用技巧
- 2022-11-02 Android?shape標簽使用方法介紹_Android
- 最近更新
-
- window11 系統安裝 yarn
- 超詳細win安裝深度學習環境2025年最新版（
- Linux 中運行的top命令 怎么退出？
- MySQL 中decimal 的用法？ 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎操作-- 運算符，流程控制 Flo
- 1. Int 和Integer 的區別，Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權
- redisson分布式鎖中waittime的設
- maven:解決release錯誤：Artif
- restTemplate使用總結
- Spring Security之安全異常處理
- MybatisPlus優雅實現加密？
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務發現-Nac
- Spring Security之基于HttpR
- Redis 底層數據結構-簡單動態字符串（SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應用詳解
- 聊聊消息隊列，發送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支
提供CDN加速