本文目录导读：
在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使用率、内存使用率、处理时间、队列长度等
- 调整策略：比例调整、线性调整、指数调整
- 调整频率：每批处理完后调整、定时调整、阈值触发调整
- 边界控制：设置最小/最大批次大小，避免过度调整
选择哪种方案取决于你的具体场景和数据特点。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
