Python脚本如何统计分布式同步整体指标

本文目录导读：

1. 核心指标定义
2. 分布式指标采集方案
3. 时间窗口统计
4. 完整统计脚本示例
5. 最佳实践建议

要统计分布式同步的整体指标，需要结合分布式系统的特点进行设计,以下是几种常见的方法和实现思路：

核心指标定义

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()

最佳实践建议

1. 时序数据库集成：将指标存储到 InfluxDB、Prometheus 等时序数据库
2. 可视化：结合 Grafana 进行展示
3. 告警机制：设置阈值，邮件/钉钉/企业微信通知
4. 采样策略：根据系统规模调整采样频率和粒度
5. 容错处理：当某个节点不可达时使用指数退避重试

这个方案可以根据实际需求进行调整和扩展。