Python脚本如何迭代优化同步执行逻辑

wen python案例 2026-06-25 252

本文目录导读：

基础同步执行框架
带超时控制的同步执行
迭代优化版本 - 动态调整执行顺序
批量处理优化
完整的迭代优化示例
关键优化策略

我来详细介绍Python中迭代优化同步执行逻辑的几种方法：

基础同步执行框架

import time
from typing import List, Callable
class SyncOptimizer:
    def __init__(self, tasks: List[Callable], max_iterations: int = 10):
        self.tasks = tasks
        self.max_iterations = max_iterations
        self.results = []
    def execute_sync(self):
        """基础同步执行"""
        for task in self.tasks:
            start = time.time()
            result = task()
            elapsed = time.time() - start
            self.results.append({
                'task': task.__name__,
                'result': result,
                'duration': elapsed
            })
        return self.results

带超时控制的同步执行

import signal
from functools import wraps
class TimeoutError(Exception):
    pass
def timeout_handler(seconds):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            def _handler(signum, frame):
                raise TimeoutError(f"Task {func.__name__} timed out after {seconds}s")
            signal.signal(signal.SIGALRM, _handler)
            signal.alarm(seconds)
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)
            return result
        return wrapper
    return decorator
class TimeoutSyncExecutor:
    def __init__(self, tasks, timeout=10):
        self.tasks = tasks
        self.timeout = timeout
    def execute_with_timeout(self):
        results = []
        for task in self.tasks:
            try:
                result = timeout_handler(self.timeout)(task)()
                results.append({'status': 'success', 'result': result})
            except TimeoutError as e:
                results.append({'status': 'timeout', 'error': str(e)})
        return results

迭代优化版本 - 动态调整执行顺序

class AdaptiveSyncOptimizer:
    def __init__(self, tasks: List[Callable], convergence_threshold=0.1):
        self.tasks = tasks
        self.convergence_threshold = convergence_threshold
        self.execution_history = {}
    def optimize_execution_order(self):
        """基于历史执行时间优化任务执行顺序"""
        for task in self.tasks:
            task_name = task.__name__
            if task_name not in self.execution_history:
                self.execution_history[task_name] = []
        # 按平均执行时间排序（短任务优先）
        sorted_tasks = sorted(
            self.tasks,
            key=lambda t: self._get_average_time(t.__name__)
        )
        return sorted_tasks
    def _get_average_time(self, task_name):
        history = self.execution_history.get(task_name, [])
        if not history:
            return float('inf')  # 新任务优先执行
        return sum(history) / len(history)
    def execute(self):
        """执行并优化"""
        for iteration in range(10):  # 迭代优化
            print(f"\n--- Iteration {iteration + 1} ---")
            # 动态优化执行顺序
            optimized_tasks = self.optimize_execution_order()
            iteration_results = []
            for task in optimized_tasks:
                start = time.time()
                result = task()
                elapsed = time.time() - start
                # 更新历史记录
                task_name = task.__name__
                self.execution_history[task_name].append(elapsed)
                iteration_results.append({
                    'task': task_name,
                    'result': result,
                    'duration': elapsed
                })
            # 检查收敛条件
            if self._check_convergence(iteration_results):
                print("Convergence achieved!")
                break
        return iteration_results
    def _check_convergence(self, results):
        """检查执行时间是否收敛"""
        durations = [r['duration'] for r in results]
        if len(durations) < 2:
            return False
        mean = sum(durations) / len(durations)
        variance = sum((d - mean) ** 2 for d in durations) / len(durations)
        return variance < self.convergence_threshold

批量处理优化

class BatchSyncExecutor:
    def __init__(self, tasks: List[Callable], batch_size=5):
        self.tasks = tasks
        self.batch_size = batch_size
        self.performance_metrics = {}
    def execute_in_batches(self):
        """分批次执行，并动态调整批次大小"""
        total_tasks = len(self.tasks)
        results = []
        for batch_start in range(0, total_tasks, self.batch_size):
            batch = self.tasks[batch_start:batch_start + self.batch_size]
            # 执行批次中的任务
            batch_results = []
            for task in batch:
                start = time.time()
                result = task()
                duration = time.time() - start
                batch_results.append({
                    'task_name': task.__name__,
                    'result': result,
                    'duration': duration
                })
                # 更新性能指标
                self._update_metrics(task.__name__, duration)
            results.extend(batch_results)
            # 动态调整批次大小
            self._adjust_batch_size()
        return results
    def _update_metrics(self, task_name, duration):
        if task_name not in self.performance_metrics:
            self.performance_metrics[task_name] = []
        self.performance_metrics[task_name].append(duration)
    def _adjust_batch_size(self):
        """根据最近的性能数据调整批次大小"""
        if len(self.performance_metrics) < 2:
            return
        avg_duration = sum(
            sum(v) / len(v) 
            for v in self.performance_metrics.values()
        ) / len(self.performance_metrics)
        self.batch_size = max(1, min(10, int(self.batch_size * (1 / avg_duration))))

完整的迭代优化示例

import random
# 模拟任务
def task_a():
    time.sleep(random.uniform(0.1, 0.5))
    return "A completed"
def task_b():
    time.sleep(random.uniform(0.2, 0.6))
    return "B completed"
def task_c():
    time.sleep(random.uniform(0.3, 0.7))
    return "C completed"
def main():
    # 使用自适应优化器
    optimizer = AdaptiveSyncOptimizer(
        tasks=[task_a, task_b, task_c],
        convergence_threshold=0.05
    )
    results = optimizer.execute()
    print("\nFinal Results:")
    for r in results:
        print(f"{r['task']}: {r['duration']:.3f}s -> {r['result']}")
    # 统计信息
    total_time = sum(r['duration'] for r in results)
    print(f"\nTotal time: {total_time:.3f}s")
    # 优化后的执行顺序
    optimized_order = optimizer.optimize_execution_order()
    print("Optimized order:", [t.__name__ for t in optimized_order])
if __name__ == "__main__":
    main()