本文目录导读：
要统计分布式同步的整体指标，需要结合分布式系统的特点进行设计,以下是几种常见的方法和实现思路：
核心指标定义
from dataclasses import dataclass
from typing import Dict, List
import time
@dataclass
class SyncMetrics:
total_nodes: int
active_nodes: int
sync_completed: int
sync_failed: int
avg_sync_time: float
max_sync_time: float
min_sync_time: float
data_volume_mb: float
sync_rate: float # 成功/总请求
latency_p99: float
latency_p95: float
分布式指标采集方案
中心化采集器
import asyncio
import aiohttp
from collections import defaultdict
class CentralizedMetricCollector:
def __init__(self, config_path='nodes_config.yaml'):
self.nodes = self.load_nodes(config_path)
self.metrics_buffer = defaultdict(list)
async def collect_from_node(self, node_url):
"""从单个节点采集指标"""
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{node_url}/metrics",
timeout=aiohttp.ClientTimeout(total=5)
) as response:
metrics = await response.json()
return metrics
except Exception as e:
print(f"Failed to collect from {node_url}: {e}")
return None
async def collect_all_metrics(self):
"""并行采集所有节点指标"""
tasks = [self.collect_from_node(node['url'])
for node in self.nodes]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [r for r in results if r is not None]
def aggregate_metrics(self, all_metrics: List[Dict]):
"""聚合所有节点指标"""
aggregated = {
'total_nodes': len(all_metrics),
'total_sync_count': sum(m.get('sync_count', 0) for m in all_metrics),
'total_sync_errors': sum(m.get('error_count', 0) for m in all_metrics),
'avg_response_time': sum(m.get('avg_time', 0) for m in all_metrics) / len(all_metrics),
'total_data_volume': sum(m.get('data_volume', 0) for m in all_metrics)
}
# 计算百分位延迟
all_latencies = []
for m in all_metrics:
all_latencies.extend(m.get('latency_samples', []))
if all_latencies:
all_latencies.sort()
n = len(all_latencies)
aggregated['p95_latency'] = all_latencies[int(n * 0.95)]
aggregated['p99_latency'] = all_latencies[int(n * 0.99)]
# 计算同步成功率
success = aggregated['total_sync_count'] - aggregated['total_sync_errors']
aggregated['sync_success_rate'] = success / max(aggregated['total_sync_count'], 1) * 100
return aggregated
def load_nodes(self, config_path):
import yaml
with open(config_path, 'r') as f:
config = yaml.safe_load(f)
return config['nodes']
使用消息队列（推荐）
import json
import redis
from typing import Dict, Any
class MessageQueueCollector:
def __init__(self, redis_host='localhost', redis_port=6379):
self.redis_client = redis.Redis(
host=redis_host,
port=redis_port,
decode_responses=True
)
self.metrics_key = 'distributed:sync:metrics'
def push_metrics(self, node_id: str, metrics: Dict[str, Any]):
"""节点推送指标到消息队列"""
message = {
'node_id': node_id,
'timestamp': time.time(),
'metrics': metrics
}
self.redis_client.lpush(self.metrics_key, json.dumps(message))
def pull_metrics(self, batch_size=100) -> List[Dict]:
"""从消息队列拉取指标"""
messages = []
for _ in range(batch_size):
msg = self.redis_client.rpop(self.metrics_key)
if msg:
messages.append(json.loads(msg))
else:
break
return messages
def get_real_time_stats(self) -> Dict:
"""获取实时统计"""
recent_metrics = self.pull_metrics(batch_size=1000)
if not recent_metrics:
return {}
# 按节点分组
node_metrics = defaultdict(list)
for msg in recent_metrics:
node_metrics[msg['node_id']].append(msg['metrics'])
# 计算每个节点的统计
node_stats = {}
for node_id, metrics_list in node_metrics.items():
node_stats[node_id] = self.calculate_node_stats(metrics_list)
return {
'node_count': len(node_stats),
'nodes': node_stats,
'overall': self.calculate_overall_stats(node_stats)
}
def calculate_node_stats(self, metrics_list: List[Dict]) -> Dict:
"""计算单个节点统计"""
sync_times = [m.get('sync_time', 0) for m in metrics_list]
return {
'total_syncs': len(metrics_list),
'avg_sync_time': sum(sync_times) / len(sync_times) if sync_times else 0,
'max_sync_time': max(sync_times),
'min_sync_time': min(sync_times)
}
def calculate_overall_stats(self, node_stats: Dict[str, Dict]) -> Dict:
"""计算整体统计"""
total_syncs = sum(s['total_syncs'] for s in node_stats.values())
avg_times = [s['avg_sync_time'] for s in node_stats.values()]
return {
'total_syncs': total_syncs,
'avg_sync_time': sum(avg_times) / len(avg_times) if avg_times else 0,
'nodes_active': len(node_stats)
}
时间窗口统计
from collections import deque
import threading
import time
class TimeWindowMetrics:
def __init__(self, window_size=60): # 60秒窗口
self.window = deque(maxlen=window_size)
self.lock = threading.Lock()
self.current_second = {}
def record_event(self, event_type: str, value: float = 1.0):
"""记录事件"""
with self.lock:
current_time = int(time.time())
# 每秒聚合
if current_time not in self.current_second:
self.current_second[current_time] = {
'count': 0,
'total_value': 0,
'events': defaultdict(int)
}
sec_data = self.current_second[current_time]
sec_data['count'] += 1
sec_data['total_value'] += value
sec_data['events'][event_type] += 1
# 移除非当前秒数据到窗口
self.flush_old_seconds(current_time)
def flush_old_seconds(self, current_time: int):
"""将超过1秒的数据移到窗口"""
to_remove = []
for sec in self.current_second:
if sec < current_time:
self.window.append({
'timestamp': sec,
**self.current_second[sec]
})
to_remove.append(sec)
for sec in to_remove:
del self.current_second[sec]
def get_window_stats(self) -> Dict:
"""获取窗口统计"""
with self.lock:
if not self.window:
return {}
total_events = sum(w['count'] for w in self.window)
total_value = sum(w['total_value'] for w in self.window)
window_duration = len(self.window)
# 事件类型统计
event_stats = defaultdict(int)
for w in self.window:
for event_type, count in w['events'].items():
event_stats[event_type] += count
return {
'events_per_second': total_events / window_duration if window_duration else 0,
'avg_value': total_value / total_events if total_events else 0,
'event_distribution': dict(event_stats),
'window_duration_seconds': window_duration
}
完整统计脚本示例
#!/usr/bin/env python3
"""
分布式同步统计脚本
支持多种采集方式：Redis/HTTP/数据库
"""
import sys
import json
import time
from datetime import datetime
import argparse
from typing import Dict, List, Optional
class DistributedSyncStats:
def __init__(self, collector_type='redis', **kwargs):
self.collector_type = collector_type
self.collector = self.init_collector(kwargs)
self.metrics_history = []
def init_collector(self, config):
"""初始化采集器"""
if self.collector_type == 'redis':
return MessageQueueCollector(
config.get('redis_host', 'localhost'),
config.get('redis_port', 6379)
)
elif self.collector_type == 'http':
return CentralizedMetricCollector(
config.get('config_path', 'nodes.yaml')
)
else:
raise ValueError(f"Unsupported collector type: {self.collector_type}")
def collect_and_report(self):
"""采集并报告"""
print(f"\n=== 分布式同步统计报告 ===")
print(f"时间: {datetime.now().isoformat()}")
if self.collector_type == 'redis':
stats = self.collector.get_real_time_stats()
else:
metrics = asyncio.run(self.collector.collect_all_metrics())
stats = self.collector.aggregate_metrics(metrics)
self.pretty_print(stats)
return stats
def pretty_print(self, stats: Dict):
"""美化输出"""
print(f"\n📊 整体指标:")
print(f" - 总节点数: {stats.get('total_nodes', stats.get('node_count', 'N/A'))}")
print(f" - 活跃节点: {stats.get('active_nodes', stats.get('nodes_active', 'N/A'))}")
print(f" - 同步总量: {stats.get('total_sync_count', stats.get('total_syncs', 'N/A'))}")
print(f" - 成功数: {stats.get('success_count', 'N/A')}")
print(f" - 失败数: {stats.get('error_count', stats.get('sync_failed', 'N/A'))}")
print(f" - 成功率: {stats.get('sync_success_rate', 'N/A')}%")
print(f" - 平均延迟: {stats.get('avg_response_time', stats.get('avg_sync_time', 'N/A'))}ms")
print(f" - P95延迟: {stats.get('p95_latency', 'N/A')}ms")
print(f" - P99延迟: {stats.get('p99_latency', 'N/A')}ms")
print(f" - 总数据量: {stats.get('total_data_volume', 'N/A')}MB")
# 如果有节点级别详情
if 'nodes' in stats:
print(f"\n📈 节点详情:")
for node_id, node_stat in stats['nodes'].items():
print(f" [{node_id}]: sync={node_stat.get('total_syncs', 0)}, "
f"avg_time={node_stat.get('avg_sync_time', 0):.2f}ms")
def main():
parser = argparse.ArgumentParser(description='分布式同步统计工具')
parser.add_argument('--type', choices=['redis', 'http'], default='redis',
help='采集器类型')
parser.add_argument('--interval', type=int, default=10,
help='采集间隔（秒）')
parser.add_argument('--redis-host', default='localhost', help='Redis地址')
parser.add_argument('--redis-port', type=int, default=6379, help='Redis端口')
parser.add_argument('--continuous', action='store_true',
help='持续监控模式')
args = parser.parse_args()
collector_config = {
'redis_host': args.redis_host,
'redis_port': args.redis_port
}
stats = DistributedSyncStats(args.type, **collector_config)
if args.continuous:
try:
while True:
stats.collect_and_report()
print(f"\n等待 {args.interval} 秒后再次采集...")
time.sleep(args.interval)
except KeyboardInterrupt:
print("\n\n采集停止")
else:
stats.collect_and_report()
if __name__ == '__main__':
main()
最佳实践建议
- 时序数据库集成：将指标存储到 InfluxDB、Prometheus 等时序数据库
- 可视化：结合 Grafana 进行展示
- 告警机制：设置阈值，邮件/钉钉/企业微信通知
- 采样策略：根据系统规模调整采样频率和粒度
- 容错处理：当某个节点不可达时使用指数退避重试
这个方案可以根据实际需求进行调整和扩展。
