本文目录导读：
我来详细介绍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()
关键优化策略
- 任务排序优化：根据历史执行时间动态调整执行顺序
- 超时控制：防止单个任务阻塞整个流程
- 批量处理：合理划分批次，平衡资源利用
- 收敛检测：当优化效果不明显时停止迭代
- 性能监控：持续追踪每个任务的执行时间
这种迭代优化方法可以在不引入异步编程的情况下,显著提升同步代码的执行效率。
