Python APScheduler 定时任务调度教程：实现高效任务管理的最佳实践-有勇气的牛排个人博客

1.1 APScheduler 基本概念

Trigger : 决定任务何时执行的触发器。APScheduler 支持多种触发器，例如


     date

、


     interval

和


     cron

。

Scheduler : 任务调度器，负责管理和执行任务。

apscheduler==3.10.4
测试环境：Python 3.10.9
3 APScheduler 触发器类型

3.1 指定时间 ymd hms

scheduler.add_job(func, 'date', run_date='2024-11-01 10:00:00')
3.2 固定时间 间隔 执行任务

# 每60秒执行一次
scheduler.add_job(func, 'interval', seconds=60)  
3.3 cron 表达式

# 每天 19:00 执行一次
scheduler.add_job(func, 'cron', hour=19, minute=0)
# 每周一至周五的 6:30 执行
scheduler.add_job(my_job, 'cron', day_of_week='mon-fri', hour=6, minute=30)  
4 基本示例: 每2s执行一次

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
# 初始化調度器
scheduler = BlockingScheduler()
def my_job():
    print("任务运行时间:", datetime.now())
def start_cron():
    # 每 2s 運行一次
    scheduler.add_job(my_job, "interval", seconds=2)
    # 启动调度器
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
if __name__ == '__main__':
    start_cron()
5 进阶功能

APScheduler 支持任务参数传递、任务暂停与恢复、错误处理和多种调度器类型。下面是一些进阶功能。
5.1 任务参数传递

def print_msg(msg, url):
    print(f"{msg}: {url}")
# 每 1s 運行一次
scheduler.add_job(print_msg, "interval", seconds=1, args=['有勇氣的牛排定時任務', "https://www.couragesteak.com"])
5.2 任务暂停与恢复

job = scheduler.add_job(print_msg, 'interval', seconds=5, args=['有勇氣的牛排定時任務'])
job.pause()  # 暂停任务
job.resume()  # 恢复任务
5.3 错误处理与任务日志

可以使用 job_defaults 配置任务出错后的重试次数、任务日志等。
scheduler.configure(job_defaults={'misfire_grace_time': 30})
6 协成异步支持

import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from datetime import datetime
# 定义一个异步任务
async def print_msg(msg, url):
    print(f"{datetime.now()} - {msg}: {url}")
    # 模拟异步操作
    await asyncio.sleep(1)
    print("异步任务执行完成")
# 启动调度器并添加异步任务
def start_async_cron():
    scheduler = AsyncIOScheduler()
    # 每 5 秒运行一次异步任务
    scheduler.add_job(
        print_msg,
        'interval',
        seconds=1,
        args=['有勇氣的牛排定時任務', "https://www.couragesteak.com"],
        max_instances=3  # 允许最多 3 个实例同时运行
    # 启动调度器
    scheduler.start()
    # 启动 asyncio 事件循环
    try:
        asyncio.get_event_loop().run_forever()
    except (KeyboardInterrupt, SystemExit):
        print("正在关闭调度器...")
        scheduler.shutdown()
        print("调度器已关闭")
if __name__ == '__main__':
    start_async_cron()
                <h2><a id="1__0"></a>1 前言</h2>
<p><code>APScheduler</code> (Advanced Python Scheduler) 是一个 Python 库，用于在指定的时间执行调度任务。它支持多种调度方式，包括固定间隔、固定日期和时间间隔等。<code>APScheduler</code> 常用于 Web 开发、数据定时采集、任务自动化等场景。</p>
<h3><a id="11_APScheduler__4"></a>1.1 APScheduler 基本概念</h3>
<li><strong>Job</strong>: 一个任务，即需要定时执行的函数。</li>
<li><strong>Trigger</strong>: 决定任务何时执行的触发器。APScheduler 支持多种触发器，例如 <code>date</code>、<code>interval</code> 和 <code>cron</code>。</li>
<li><strong>Scheduler</strong>: 任务调度器，负责管理和执行任务。</li>
<h2><a id="2__12"></a>2 安装</h2>
<pre><div class="hljs"><code class="lang-shell">apscheduler==3.10.4
</code></div></pre>
<p>测试环境：Python 3.10.9</p>
<h2><a id="3_APScheduler__20"></a>3 APScheduler 触发器类型</h2>
<h3><a id="31__ymd_hms_22"></a>3.1 指定时间 ymd hms</h3>
<pre><div class="hljs"><code class="lang-python">scheduler.add_job(func, <span class="hljs-string">&#x27;date&#x27;</span>, run_date=<span class="hljs-string">&#x27;2024-11-01 10:00:00&#x27;</span>)
</code></div></pre>
<h3><a id="32____28"></a>3.2 固定时间 间隔 执行任务</h3>
<pre><div class="hljs"><code class="lang-python"><span class="hljs-comment"># 每60秒执行一次</span>
scheduler.add_job(func, <span class="hljs-string">&#x27;interval&#x27;</span>, seconds=<span class="hljs-number">60</span>)  
</code></div></pre>
<h3><a id="33_cron__35"></a>3.3 cron 表达式</h3>
<pre><div class="hljs"><code class="lang-python"><span class="hljs-comment"># 每天 19:00 执行一次</span>
scheduler.add_job(func, <span class="hljs-string">&#x27;cron&#x27;</span>, hour=<span class="hljs-number">19</span>, minute=<span class="hljs-number">0</span>)
</code></div></pre>
<pre><div class="hljs"><code class="lang-python"><span class="hljs-comment"># 每周一至周五的 6:30 执行</span>
scheduler.add_job(my_job, <span class="hljs-string">&#x27;cron&#x27;</span>, day_of_week=<span class="hljs-string">&#x27;mon-fri&#x27;</span>, hour=<span class="hljs-number">6</span>, minute=<span class="hljs-number">30</span>)  
</code></div></pre>
<h2><a id="4__2s_49"></a>4 基本示例: 每2s执行一次</h2>
<pre><div class="hljs"><code class="lang-python"><span class="hljs-keyword">from</span> apscheduler.schedulers.blocking <span class="hljs-keyword">import</span> BlockingScheduler
<span class="hljs-keyword">from</span> datetime <span class="hljs-keyword">import</span> datetime
<span class="hljs-comment"># 初始化調度器</span>
scheduler = BlockingScheduler()
<span class="hljs-keyword">def</span> <span class="hljs-title function_">my_job</span>():
    <span class="hljs-built_in">print</span>(<span class="hljs-string">&quot;任务运行时间:&quot;</span>, datetime.now())
<span class="hljs-keyword">def</span> <span class="hljs-title function_">start_cron</span>():
    <span class="hljs-comment"># 每 2s 運行一次</span>
    scheduler.add_job(my_job, <span class="hljs-string">&quot;interval&quot;</span>, seconds=<span class="hljs-number">2</span>)
    <span class="hljs-comment"># 启动调度器</span>
    <span class="hljs-keyword">try</span>:
        scheduler.start()
    <span class="hljs-keyword">except</span> (KeyboardInterrupt, SystemExit):
        <span class="hljs-keyword">pass</span>
<span class="hljs-keyword">if</span> __name__ == <span class="hljs-string">&#x27;__main__&#x27;</span>:
    start_cron()
</code></div></pre>
<p><img src="https://static.couragesteak.com/article/64747cb756f4d49c1895d30c4823feb5.png" alt="Python定时任务 基本示例: 每2s执行一次" /></p>
<h2><a id="5__80"></a>5 进阶功能</h2>
<p><code>APScheduler</code> 支持任务参数传递、任务暂停与恢复、错误处理和多种调度器类型。下面是一些进阶功能。</p>
<h3><a id="51__84"></a>5.1 任务参数传递</h3>
<pre><div class="hljs"><code class="lang-python"><span class="hljs-keyword">def</span> <span class="hljs-title function_">print_msg</span>(<span class="hljs-params">msg, url</span>):
    <span class="hljs-built_in">print</span>(<span class="hljs-string">f&quot;<span class="hljs-subst">{msg}</span>: <span class="hljs-subst">{url}</span>&quot;</span>)
<span class="hljs-comment"># 每 1s 運行一次</span>
scheduler.add_job(print_msg, <span class="hljs-string">&quot;interval&quot;</span>, seconds=<span class="hljs-number">1</span>, args=[<span class="hljs-string">&#x27;有勇氣的牛排定時任務&#x27;</span>, <span class="hljs-string">&quot;https://www.couragesteak.com&quot;</span>])
</code></div></pre>
<p><img src="https://static.couragesteak.com/article/65d4c5d806a9dbf962a582d4e32c8057.png" alt="Python定时任务 任务参数传递" /></p>
<h3><a id="52__96"></a>5.2 任务暂停与恢复</h3>
<pre><div class="hljs"><code class="lang-python">job = scheduler.add_job(print_msg, <span class="hljs-string">&#x27;interval&#x27;</span>, seconds=<span class="hljs-number">5</span>, args=[<span class="hljs-string">&#x27;有勇氣的牛排定時任務&#x27;</span>])
job.pause()  <span class="hljs-comment"># 暂停任务</span>
job.resume()  <span class="hljs-comment"># 恢复任务</span>
</code></div></pre>
<h3><a id="53__104"></a>5.3 错误处理与任务日志</h3>
<p>可以使用 <code>job_defaults</code> 配置任务出错后的重试次数、任务日志等。</p>
<pre><div class="hljs"><code class="lang-python">scheduler.configure(job_defaults={<span class="hljs-string">&#x27;misfire_grace_time&#x27;</span>: <span class="hljs-number">30</span>})
</code></div></pre>
<h2><a id="6__112"></a>6 协成异步支持</h2>
<pre><div class="hljs"><code class="lang-python"><span class="hljs-keyword">import</span> asyncio
<span class="hljs-keyword">from</span> apscheduler.schedulers.asyncio <span class="hljs-keyword">import</span> AsyncIOScheduler
<span class="hljs-keyword">from</span> datetime <span class="hljs-keyword">import</span> datetime
<span class="hljs-comment"># 定义一个异步任务</span>
<span class="hljs-keyword">async</span> <span class="hljs-keyword">def</span> <span class="hljs-title function_">print_msg</span>(<span class="hljs-params">msg, url</span>):
    <span class="hljs-built_in">print</span>(<span class="hljs-string">f&quot;<span class="hljs-subst">{datetime.now()}</span> - <span class="hljs-subst">{msg}</span>: <span class="hljs-subst">{url}</span>&quot;</span>)
    <span class="hljs-comment"># 模拟异步操作</span>
    <span class="hljs-keyword">await</span> asyncio.sleep(<span class="hljs-number">1</span>)
    <span class="hljs-built_in">print</span>(<span class="hljs-string">&quot;异步任务执行完成&quot;</span>)
<span class="hljs-comment"># 启动调度器并添加异步任务</span>
<span class="hljs-keyword">def</span> <span class="hljs-title function_">start_async_cron</span>():
    scheduler = AsyncIOScheduler()
    <span class="hljs-comment"># 每 5 秒运行一次异步任务</span>
    scheduler.add_job(
        print_msg,
        <span class="hljs-string">&#x27;interval&#x27;</span>,
        seconds=<span class="hljs-number">1</span>,
        args=[<span class="hljs-string">&#x27;有勇氣的牛排定時任務&#x27;</span>, <span class="hljs-string">&quot;https://www.couragesteak.com&quot;</span>],
        max_instances=<span class="hljs-number">3</span>  <span class="hljs-comment"># 允许最多 3 个实例同时运行</span>
    <span class="hljs-comment"># 启动调度器</span>
    scheduler.start()
    <span class="hljs-comment"># 启动 asyncio 事件循环</span>
    <span class="hljs-keyword">try</span>:
        asyncio.get_event_loop().run_forever()
    <span class="hljs-keyword">except</span> (KeyboardInterrupt, SystemExit):
        <span class="hljs-built_in">print</span>(<span class="hljs-string">&quot;正在关闭调度器...&quot;</span>)
        scheduler.shutdown()
        <span class="hljs-built_in">print</span>(<span class="hljs-string">&quot;调度器已关闭&quot;</span>)
<span class="hljs-keyword">if</span> __name__ == <span class="hljs-string">&#x27;__main__&#x27;</span>:
    start_async_cron()
</code></div></pre>
<p><img src="https://static.couragesteak.com/article/a4620bbbc9dfdc6dd7a9d877bc1711a5.png" alt="Python定时任务 协成异步支持" /></p>