本文目录导读：
在Python中动态调整同步批次数量（batch size）主要有以下几种实现方式：
基于系统资源动态调整
import psutil
import time
from typing import List, Any
class DynamicBatchProcessor:
def __init__(self,
min_batch: int = 1,
max_batch: int = 100,
target_memory_usage: float = 0.7):
"""
min_batch: 最小批次大小
max_batch: 最大批次大小
target_memory_usage: 目标内存使用率 (0-1)
"""
self.current_batch = min_batch
self.min_batch = min_batch
self.max_batch = max_batch
self.target_memory_usage = target_memory_usage
def calculate_optimal_batch(self) -> int:
"""根据当前系统资源计算最优批次大小"""
memory_percent = psutil.virtual_memory().percent / 100
if memory_percent > self.target_memory_usage:
# 内存使用过高，减少批次大小
self.current_batch = max(
self.min_batch,
int(self.current_batch * 0.8)
)
else:
# 内存充足，增加批次大小
self.current_batch = min(
self.max_batch,
int(self.current_batch * 1.2) + 1
)
return self.current_batch
def process_batch(self, items: List[Any]) -> None:
"""处理数据的批次"""
batch_size = self.calculate_optimal_batch()
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
self._process_single_batch(batch)
# 处理完每批后重新评估
time.sleep(0.1) # 模拟处理时间
self.calculate_optimal_batch()
def _process_single_batch(self, batch: List[Any]) -> None:
"""处理单个批次的逻辑"""
# 这里实现你的业务逻辑
pass
基于处理时间动态调整
import time
from collections import deque
from typing import List, Any
class TimeBasedBatchAdjuster:
def __init__(self,
initial_batch: int = 32,
target_time: float = 1.0, # 目标处理时间（秒）
tolerance: float = 0.2): # 容忍度
self.batch_size = initial_batch
self.target_time = target_time
self.tolerance = tolerance
self.processing_times = deque(maxlen=10) # 最近10次处理时间
def process(self, data: List[Any]) -> None:
"""处理数据并动态调整batch size"""
idx = 0
while idx < len(data):
batch = data[idx:idx + self.batch_size]
# 记录处理时间
start_time = time.time()
self._execute_batch(batch)
elapsed_time = time.time() - start_time
self.processing_times.append(elapsed_time)
self._adjust_batch_size()
idx += self.batch_size
def _adjust_batch_size(self) -> None:
"""根据处理时间调整批次大小"""
if len(self.processing_times) < 5:
return # 样本不够，暂不调整
avg_time = sum(self.processing_times) / len(self.processing_times)
if avg_time < self.target_time * (1 - self.tolerance):
# 处理太快，增大批次
self.batch_size = int(self.batch_size * 1.5)
elif avg_time > self.target_time * (1 + self.tolerance):
# 处理太慢，减小批次
self.batch_size = max(1, int(self.batch_size * 0.7))
def _execute_batch(self, batch: List[Any]) -> None:
"""执行批次处理逻辑"""
# 实现你的实际处理逻辑
pass
综合控制方案
import psutil
import time
from typing import List, Any, Callable
from dataclasses import dataclass
@dataclass
class BatchConfig:
"""批次配置"""
current: int = 32
min_size: int = 1
max_size: int = 256
adjustment_factor: float = 0.1
class AdaptiveBatchController:
def __init__(self,
config: BatchConfig = None,
monitor_interval: float = 1.0):
self.config = config or BatchConfig()
self.monitor_interval = monitor_interval
self.metrics = {
'cpu_percent': 0,
'memory_percent': 0,
'avg_processing_time': 0
}
def process_data(self,
data: List[Any],
processing_func: Callable) -> List[Any]:
"""处理数据并自适应调整batch size"""
results = []
idx = 0
while idx < len(data):
# 动态计算批次大小
batch_size = self._get_adaptive_batch_size()
# 获取当前批次
batch = data[idx:idx + batch_size]
# 处理批次
start_time = time.time()
batch_results = processing_func(batch)
processing_time = time.time() - start_time
# 更新指标
self._update_metrics(processing_time)
results.extend(batch_results)
idx += batch_size
# 调整批次大小
self._adjust_batch_size(processing_time)
return results
def _get_adaptive_batch_size(self) -> int:
"""获取自适应批次大小"""
return self.config.current
def _update_metrics(self, processing_time: float) -> None:
"""更新系统指标"""
self.metrics['cpu_percent'] = psutil.cpu_percent()
self.metrics['memory_percent'] = psutil.virtual_memory().percent
self.metrics['avg_processing_time'] = (
0.9 * self.metrics['avg_processing_time'] +
0.1 * processing_time
)
def _adjust_batch_size(self, processing_time: float) -> None:
"""调整批次大小"""
# 基于处理时间调整
if processing_time > 2.0: # 处理时间过长
self.config.current = max(
self.config.min_size,
int(self.config.current * (1 - self.config.adjustment_factor))
)
elif processing_time < 0.5: # 处理时间过短
self.config.current = min(
self.config.max_size,
int(self.config.current * (1 + self.config.adjustment_factor))
)
# 基于内存使用调整
if self.metrics['memory_percent'] > 80: # 内存使用超过80%
self.config.current = max(
self.config.min_size,
int(self.config.current * 0.8)
)
def get_current_metrics(self) -> dict:
"""获取当前指标"""
return self.metrics
# 使用示例
def example_processing_function(batch):
"""示例处理函数"""
return [item * 2 for item in batch]
# 创建控制器
controller = AdaptiveBatchController(
config=BatchConfig(
current=50,
min_size=10,
max_size=200
)
)
# 处理数据
data = list(range(1000))
results = controller.process_data(
data,
example_processing_function
)
基于队列的异步方案（适合生产者-消费者模式）
import queue
import threading
import time
from typing import Any, Callable
class DynamicBatchQueue:
def __init__(self,
processing_func: Callable,
initial_batch: int = 32):
self.input_queue = queue.Queue()
self.output_queue = queue.Queue()
self.batch_size = initial_batch
self.processing_func = processing_func
self.running = False
def start(self, num_workers: int = 4):
"""启动工作者线程"""
self.running = True
self.workers = []
for _ in range(num_workers):
worker = threading.Thread(target=self._worker_loop)
worker.daemon = True
worker.start()
self.workers.append(worker)
def stop(self):
"""停止所有工作者"""
self.running = False
for worker in self.workers:
worker.join()
def _worker_loop(self):
"""工作者主循环"""
while self.running:
# 收集批次数据
batch = []
start_time = time.time()
# 尝试收集足够的数据或超时
while len(batch) < self.batch_size:
try:
item = self.input_queue.get(timeout=0.1)
batch.append(item)
except queue.Empty:
if batch and time.time() - start_time > 1.0:
break
if not self.running and not batch:
return
if batch:
# 处理批次
start_process = time.time()
results = self.processing_func(batch)
process_time = time.time() - start_process
# 动态调整批次大小
self._adjust_batch_size(process_time)
# 输出结果
for result in results:
self.output_queue.put(result)
def _adjust_batch_size(self, process_time: float):
"""调整批次大小"""
if process_time > 2.0:
self.batch_size = max(1, self.batch_size - 5)
elif process_time < 0.5:
self.batch_size = min(256, self.batch_size + 5)
使用回调函数动态调整
from typing import List, Any, Callable, Optional
def dynamic_batch_processor(
data: List[Any],
process_func: Callable[[List[Any]], List[Any]],
initial_batch_size: int = 32,
adjust_callback: Optional[Callable[[dict], int]] = None
) -> List[Any]:
"""
动态批次处理器
adjust_callback: 接收处理统计信息的回调函数，返回新的批次大小
"""
results = []
batch_size = initial_batch_size
idx = 0
while idx < len(data):
batch = data[idx:idx + batch_size]
start_time = time.time()
batch_results = process_func(batch)
elapsed = time.time() - start_time
# 统计信息
stats = {
'batch_size': batch_size,
'processing_time': elapsed,
'items_processed': len(batch),
'memory_usage': psutil.virtual_memory().percent,
'cpu_usage': psutil.cpu_percent()
}
# 动态调整
if adjust_callback:
batch_size = adjust_callback(stats)
else:
# 默认调整策略
if elapsed > 1.0: # 处理时间过长
batch_size = max(1, int(batch_size * 0.8))
elif elapsed < 0.2: # 处理时间过短
batch_size = min(256, int(batch_size * 1.2))
results.extend(batch_results)
idx += len(batch)
return results
# 自定义调整回调
def custom_adjust_callback(stats: dict) -> int:
"""自定义批次大小调整策略"""
# 根据特定规则调整
if stats['cpu_usage'] > 90:
return max(1, stats['batch_size'] - 10)
elif stats['memory_usage'] < 50 and stats['processing_time'] < 0.5:
return min(256, stats['batch_size'] + 10)
return stats['batch_size']
# 使用示例
results = dynamic_batch_processor(
data=list(range(1000)),
process_func=lambda batch: [x * 2 for x in batch],
initial_batch_size=50,
adjust_callback=custom_adjust_callback
)
- 监测指标选择：CPU使用率、内存使用率、处理时间、队列长度等
- 调整策略：比例调整、线性调整、指数调整
- 调整频率：每批处理完后调整、定时调整、阈值触发调整
- 边界控制：设置最小/最大批次大小，避免过度调整
选择哪种方案取决于你的具体场景和数据特点。
