dotask——高效易用的任務執行框架

一、任務執行框架介紹

在一個複雜的系統中，通常存在大量的任務需要執行。這些任務包括但不限於：發送郵件、處理數據、調用服務、生成報表等。在傳統的編程模式中，我們往往需要手動編寫任務調度代碼，實現任務的調度、執行、監控等功能。這樣的方式不僅耗費時間和精力，而且容易出現錯誤和漏洞。因此，我們需要一個高效易用的任務執行框架來簡化任務調度流程，提高任務執行效率。

二、dotask特性

dotask是一個輕量級的任務執行框架，具有以下特性：

• 易於使用：提供簡單易用的API，支持一次性任務、定時任務、循環任務等多種任務類型。
• 高效可靠：採用多線程調度方式，支持任務的並發執行，避免任務阻塞，同時提供任務調度監控和異常處理機制。
• 靈活擴展：支持自定義任務執行邏輯和任務調度策略，並提供多種插件機制，可根據業務需求進行靈活擴展。

三、快速入門

下面是一個簡單的dotask任務調度示例：

from dotask import Task, TaskScheduler

# 定義任務
def send_email_task(to, content):
    # 發送郵件邏輯
    print('to:', to)
    print('content:', content)

# 創建任務
send_email_task1 = Task(send_email_task, args=('abc@example.com', 'hello world'))
send_email_task2 = Task(send_email_task, args=('123@example.com', 'dotask is cool!'))

# 創建任務調度器
scheduler = TaskScheduler()

# 添加任務
scheduler.add_task(send_email_task1)
scheduler.add_task(send_email_task2)

# 啟動任務調度器
scheduler.start()

運行結果：

to: abc@example.com
content: hello world
to: 123@example.com
content: dotask is cool!

四、任務類型

dotask支持以下任務類型：

• 一次性任務：只執行一次的任務。
• 定時任務：按照指定的時間間隔執行任務。
• 循環任務：按照指定的時間間隔循環執行任務。

下面是一個任務類型的示例：

from dotask import Task, TaskScheduler
from datetime import datetime, timedelta

# 一次性任務
onetime_task = Task(lambda: print('onetime_task'))

# 定時任務
start_time = datetime.now() + timedelta(minutes=1)
interval = timedelta(minutes=5)
interval_task = Task(lambda: print('interval_task'), start_time=start_time, interval=interval)

# 循環任務
interval_task2 = Task(lambda: print('interval_task2'), interval=timedelta(seconds=2), count=5)

# 創建任務調度器
scheduler = TaskScheduler()

# 添加任務
scheduler.add_task(onetime_task)
scheduler.add_task(interval_task)
scheduler.add_task(interval_task2)

# 啟動任務調度器
scheduler.start()

運行結果：

onetime_task
interval_task
interval_task2
interval_task2
interval_task2
interval_task2
interval_task2

五、任務執行策略

dotask提供以下任務執行策略：

• 串行執行：按照任務添加的順序串行執行任務。
• 並發執行：同時執行所有任務，無序。

下面是一個任務執行策略的示例：

from dotask import Task, TaskScheduler

# 串行執行任務
serial_task1 = Task(lambda: print('serial_task1'))
serial_task2 = Task(lambda: print('serial_task2'))

# 並發執行任務
concurrent_task1 = Task(lambda: print('concurrent_task1'))
concurrent_task2 = Task(lambda: print('concurrent_task2'))

# 創建任務調度器（並發執行）
scheduler1 = TaskScheduler(strategy='concurrent')

# 添加任務
scheduler1.add_task(concurrent_task1)
scheduler1.add_task(concurrent_task2)

# 啟動任務調度器
scheduler1.start()

# 創建任務調度器（串行執行）
scheduler2 = TaskScheduler(strategy='serial')

# 添加任務
scheduler2.add_task(serial_task1)
scheduler2.add_task(serial_task2)

# 啟動任務調度器
scheduler2.start()

運行結果：

concurrent_task1
concurrent_task2
serial_task1
serial_task2

六、插件機制

dotask提供多種插件機制，可根據業務需求進行靈活擴展。

1. 任務執行插件：根據任務類型、任務狀態等對任務進行不同的處理。例如處理一次性任務和定時任務的差異。
2. 任務調度插件：根據任務類型、任務狀態等對任務進行不同的調度策略。例如對高優先級任務的優先處理。
3. 日誌插件：記錄任務執行過程、異常信息等。
4. 監控插件：對任務執行情況進行實時監控，並提供告警和監控統計功能。

下面是一個任務執行插件的示例：

from dotask import Task, TaskScheduler, plugins

# 任務執行插件
class CustomTaskExecutor(plugins.TaskExecutePlugin):
    def execute(self, task):
        if task.type == 'onetime':
            # 處理一次性任務
            print('onetime task')
        elif task.type == 'interval':
            # 處理定時任務
            print('interval task')
        elif task.type == 'loop':
            # 處理循環任務
            print('loop task')
            if task.current_count >= task.count:
                # 執行次數達到上限，停止任務
                task.stop()

# 創建任務調度器
scheduler = TaskScheduler()

# 設置任務執行插件
scheduler.add_plugin(CustomTaskExecutor())

# 添加任務
scheduler.add_task(Task(lambda: None))
scheduler.add_task(Task(lambda: None, start_time=datetime.now() + timedelta(seconds=5), interval=timedelta(seconds=5)))
scheduler.add_task(Task(lambda: None, interval=timedelta(seconds=3), count=3))

# 啟動任務調度器
scheduler.start()

運行結果：

onetime task
loop task
loop task
loop task
interval task
loop task
loop task
interval task
loop task

七、總結

dotask是一個高效易用的任務執行框架，支持多種任務類型、任務執行策略和插件機制。通過dotask，我們可以極大地簡化任務調度的過程，提高任務的執行效率和可靠性，使開發人員可以更專註於業務邏輯的開發。