Rocketry：Python 中现代且强大的任务调度框架

前言

在现代软件开发中，任务调度是一个非常重要的环节。Rocketry 是一个基于 Python 的现代任务调度框架，它具有简单、干净和强大的特点，适用于各种规模的项目。

Rocketry 的核心功能

• 强大的调度功能：Rocketry 提供了丰富的调度功能，包括 cron 表达式、时间间隔、日期和时间等，可以满足各种复杂的任务调度需求。
• 并发支持：Rocketry 支持异步、多线程和多进程等并发模式，可以充分利用系统的资源，提高任务执行效率。
• 参数化：Rocketry 支持任务参数化，允许在任务执行时传入不同的参数，使任务更加灵活。
• 任务流水线：Rocketry 支持任务流水线，可以将多个任务串联起来执行，形成一个完整的任务流程。
• 可修改的会话：Rocketry 允许在运行时修改任务的会话，使任务能够适应动态变化的环境。
• 异步支持：Rocketry 支持异步任务，可以充分利用 Python 的异步特性，提高任务执行效率。

Rocketry 的优点

• 声明式调度：Rocketry 采用声明式调度的方式，使任务调度更加直观和易于理解。
• 可扩展性强：Rocketry 的调度器是基于语句的，可以任意扩展，支持自定义调度语句。
• 轻量级：Rocketry 非常轻量级，不会对项目带来额外的负担。

Rocketry 的安装

可以使用 pip 命令安装 Rocketry：

pip install rocketry

Rocketry 的示例

以下是一些 Rocketry 的示例：

调度任务：

# 导入 Rocketry 库
from rocketry import Rocketry

# 导入 daily 条件
from rocketry.conds import daily

# 创建 Rocketry 实例
app = Rocketry()

# 定义一个名为 do_daily 的任务
# 该任务将在每天早上 8 点后执行
@app.task(daily.after("08:00"))
def do_daily():
    # 在这里编写任务的具体逻辑

# 如果当前脚本是作为主程序运行的
if __name__ == '__main__':
    # 运行 Rocketry 实例
    app.run()

任务流水线：

# 导入 Rocketry 库
from rocketry import Rocketry

# 导入 daily 和 after_success 条件
from rocketry.conds import daily, after_success

# 导入 Return 参数
from rocketry.args import Return

# 创建 Rocketry 实例
app = Rocketry()

# 定义第一个任务 do_first
# 该任务将在每天早上 7 点后执行
@app.task(daily.after("07:00"))
def do_first():
    """
    第一个任务的具体逻辑
    """
    # 返回一个字符串
    return 'Hello World'

# 定义第二个任务 do_second
# 该任务将在第一个任务 do_first 成功执行后执行
@app.task(after_success(do_first))
def do_second(arg=Return('do_first')):
    """
    第二个任务的具体逻辑
    arg 参数包含第一个任务 do_first 的返回值
    """
    # 返回一个字符串
    return 'Hello Python'

# 如果当前脚本是作为主程序运行的
if __name__ == '__main__':
    # 运行 Rocketry 实例
    app.run()

并行任务：

# 导入 Rocketry 库
from rocketry import Rocketry

# 导入 daily 条件
from rocketry.conds import daily

# 创建 Rocketry 实例
app = Rocketry()

# 定义一个名为 do_unparallel 的任务
# 该任务将在每天执行一次，并且在主进程中执行
@app.task(daily, execution="main")
def do_unparallel():
    """
    该任务将在每天执行一次，并且在主进程中执行
    """
    ...

# 定义一个名为 do_async 的任务
# 该任务将在每天执行一次，并且以异步的方式执行
@app.task(daily, execution="async")
async def do_async():
    """
    该任务将在每天执行一次，并且以异步的方式执行
    """
    ...

# 定义一个名为 do_on_separate_thread 的任务
# 该任务将在每天执行一次，并且在一个单独的线程中执行
@app.task(daily, execution="thread")
def do_on_separate_thread():
    """
    该任务将在每天执行一次，并且在一个单独的线程中执行
    """
    ...

# 定义一个名为 do_on_separate_process 的任务
# 该任务将在每天执行一次，并且在一个单独的进程中执行
@app.task(daily, execution="process")
def do_on_separate_process():
    """
    该任务将在每天执行一次，并且在一个单独的进程中执行
    """
    ...

# 如果当前脚本是作为主程序运行的
if __name__ == '__main__':
    # 运行 Rocketry 实例
    app.run()