Python 设计模式：回调模式

1. 什么是回调函数？

回调函数是指作为参数传递给另一个函数的函数。当这个函数执行到某个特定的点时，它会调用这个回调函数。回调函数通常用于处理异步操作、事件处理或在某些条件下执行特定的操作。

回调函数的特点：

• 作为参数传递：回调函数可以作为参数传递给其他函数，这使得它们可以在需要时被调用。
• 延迟执行：回调函数的执行通常是延迟的，只有在特定条件满足时才会被调用。
• 灵活性：通过使用回调函数，开发者可以在运行时定义行为，从而使代码更加灵活和可扩展。

回调函数的应用场景：

• 事件处理：在图形用户界面（GUI）应用程序中，用户的操作（如点击按钮）可以触发回调函数。
• 异步编程：在网络请求或文件读取等异步操作中，回调函数可以在操作完成时被调用。
• 库和框架：许多库和框架使用回调函数来允许用户自定义行为，例如在数据处理或任务调度时。

def greet(name):return f"Hello, {name}!"def process_name(name, callback):# 调用回调函数result = callback(name)print(result)if __name__ == "__main__":process_name("Alice", greet)  # 将 greet 函数作为回调传递

• 定义回调函数：greet 函数是一个简单的回调函数，它接受一个名字并返回问候语。
• 处理函数：process_name 函数接受一个名字和一个回调函数作为参数。它调用回调函数并打印结果。
• 客户端代码：在客户端代码中，调用 process_name 函数，并将greet 函数作为回调传递。

2. 回调模式

回调模式是一种设计模式，其中一个函数作为参数传递给另一个函数，并在特定条件下被调用。这种模式允许程序在某些事件发生时执行特定的操作，从而实现异步编程或事件驱动编程。

回调的核心思想是将某些操作的执行延迟到后续的某个时刻。通过将函数作为参数传递，程序可以在需要时调用该函数，而不需要在调用时立即执行。这种方式使得代码更加灵活和可扩展。

回调模式的优点：

• 灵活性：回调模式允许开发者在运行时定义行为，使得代码更加灵活。
• 解耦：通过将操作的实现与调用分离，回调模式降低了模块之间的耦合度。
• 异步处理：回调模式常用于异步编程，使得程序可以在等待某些操作完成时继续执行其他任务。

3. 示例 1：任务调度系统

假设我们正在开发一个任务调度系统，该系统可以执行多个任务，并在每个任务完成后调用相应的回调函数来处理结果。我们将实现一个简单的任务调度器，支持添加任务、执行任务和处理任务结果。

import time
import random
from threading import Thread# 定义任务类
class Task:def __init__(self, name, duration, callback):self.name = nameself.duration = durationself.callback = callbackdef run(self):print(f"Task {self.name} is starting...")time.sleep(self.duration)  # 模拟耗时操作result = f"Task {self.name} completed in {self.duration} seconds."self.callback(result)  # 调用回调函数# 定义任务调度器
class TaskScheduler:def __init__(self):self.tasks = []def add_task(self, task):self.tasks.append(task)def execute_tasks(self):threads = []for task in self.tasks:thread = Thread(target=task.run)threads.append(thread)thread.start()for thread in threads:thread.join()  # 等待所有线程完成# 定义回调函数
def task_completed(result):print(f"Callback received: {result}")if __name__ == "__main__":scheduler = TaskScheduler()# 添加任务for i in range(5):duration = random.randint(1, 5)  # 随机生成任务持续时间task = Task(name=f"Task-{i + 1}", duration=duration, callback=task_completed)scheduler.add_task(task)# 执行任务scheduler.execute_tasks()

Task Task-1 is starting...Task Task-2 is starting...Task Task-3 is starting...
Task Task-4 is starting...
Task Task-5 is starting...
Callback received: Task Task-5 completed in 1 seconds.
Callback received: Task Task-1 completed in 3 seconds.
Callback received: Task Task-3 completed in 3 seconds.
Callback received: Task Task-4 completed in 5 seconds.
Callback received: Task Task-2 completed in 5 seconds.

1. 任务类：

   • Task 类表示一个任务，包含任务名称、持续时间和回调函数。
   • run 方法模拟任务的执行，使用 time.sleep 来模拟耗时操作。在任务完成后，它调用回调函数并传递结果。

2. 任务调度器：

   • TaskScheduler 类用于管理任务。它包含一个任务列表，并提供 add_task 方法来添加任务。
   • execute_tasks 方法创建一个线程来执行每个任务，并等待所有线程完成。

3. 回调函数：

   • task_completed 函数定义了在任务完成后要执行的操作。它接收任务结果并打印出来。

4. 客户端代码：

   • 在 if __name__ == "__main__": 块中，创建一个任务调度器实例。
   • 随机生成 5 个任务，并将它们添加到调度器中。
   • 调用 execute_tasks 方法执行所有任务。

示例2： GUI 编程中任务管理器应用

我们将创建一个简单的任务管理器应用，用户可以添加任务、删除任务和标记任务为完成。每当用户进行这些操作时，应用将调用相应的回调函数来更新界面。

import tkinter as tk
from tkinter import messagebox, simpledialogclass TaskManagerApp:def __init__(self, master):self.master = masterself.master.title("Task Manager")self.master.geometry("400x400")  # 设置窗口大小self.tasks = []  # 初始化任务列表# 创建任务列表框self.task_listbox = tk.Listbox(master, font=("Arial", 14))self.task_listbox.pack(pady=20, fill=tk.BOTH, expand=True)# 创建按钮框self.button_frame = tk.Frame(master)self.button_frame.pack(pady=10)# 添加任务按钮self.add_task_button = tk.Button(self.button_frame, text="Add Task", command=self.add_task)self.add_task_button.pack(side=tk.LEFT, padx=5)# 删除任务按钮self.delete_task_button = tk.Button(self.button_frame, text="Delete Task", command=self.delete_task)self.delete_task_button.pack(side=tk.LEFT, padx=5)# 标记完成按钮self.complete_task_button = tk.Button(self.button_frame, text="Complete Task", command=self.complete_task)self.complete_task_button.pack(side=tk.LEFT, padx=5)def add_task(self):task_name = simpledialog.askstring("Add Task", "Enter task name:")if task_name:self.tasks.append(task_name)self.update_task_listbox()def delete_task(self):selected_task_index = self.task_listbox.curselection()if selected_task_index:del self.tasks[selected_task_index[0]]self.update_task_listbox()else:messagebox.showwarning("Delete Task", "Please select a task to delete.")def complete_task(self):selected_task_index = self.task_listbox.curselection()if selected_task_index:task_name = self.tasks[selected_task_index[0]]messagebox.showinfo("Complete Task", f"Task '{task_name}' marked as completed.")del self.tasks[selected_task_index[0]]self.update_task_listbox()else:messagebox.showwarning("Complete Task", "Please select a task to mark as complete.")def update_task_listbox(self):self.task_listbox.delete(0, tk.END)  # 清空列表框for task in self.tasks:self.task_listbox.insert(tk.END, task)  # 添加任务到列表框if __name__ == "__main__":root = tk.Tk()  # 创建主窗口app = TaskManagerApp(root)  # 创建任务管理器应用root.mainloop()  # 运行主循环

1. 创建主窗口：

   • 使用 tk.Tk() 创建主窗口，并设置窗口标题为 “Task Manager” 和大小为 400x400 像素。

2. 任务管理器类：

   • TaskManagerApp 类负责管理任务管理器的逻辑和界面。
   • __init__ 方法初始化任务列表、创建任务列表框和按钮，并将按钮的 command 参数设置为相应的方法。

3. 添加任务：

   • add_task 方法使用 simpledialog.askstring 弹出对话框，提示用户输入任务名称。如果用户输入了任务名称，则将其添加到任务列表中，并调用 update_task_listbox 方法更新显示。

4. 删除任务：

   • delete_task 方法获取用户在列表框中选择的任务索引。如果选择了任务，则从任务列表中删除该任务，并更新列表框。如果没有选择任务，则弹出警告对话框。

5. 标记任务为完成：

   • complete_task 方法获取用户在列表框中选择的任务索引。如果选择了任务，则弹出信息对话框，显示任务已完成，并从任务列表中删除该任务。如果没有选择任务，则弹出警告对话框。

6. 更新任务列表框：

   • update_task_listbox 方法清空列表框并将当前任务列表中的所有任务添加到列表框中。
     运行上述代码时，将看到一个简单的任务管理器界面，包含一个显示任务的列表框和三个按钮（添加任务、删除任务、标记完成）。用户可以通过点击按钮来管理任务。