第三十节：综合实战——从零构建一个电源管理监控工具

终于到了最后一节。说实话，写到这里我还有点不舍。

前面29节，我们把Wakeup源、唤醒锁、suspend/resume流程、内核机制、用户空间工具全讲了一遍。但光说不练假把式。今天我们就动手，从零搭一个电源管理监控工具，把Wakeup源和唤醒锁可视化，再配上自动化测试脚本。

我在项目里做过类似的东西，当时是为了排查一个“手机待机一晚掉电15%”的线上问题。没有可视化工具，全靠dumpsys和kernel log硬啃，效率极低。后来我花了两天写了个监控脚本，问题定位时间从半天缩短到十分钟。嗯，这就是我们今天要做的。

30.1 工具架构设计

先画个整体结构图。说白了，这个工具分三层：

• 数据采集层：从/sys/kernel/debug/wakeup_sources、/d/tracing/events/power、dumpsys power等接口抓取原始数据
• 数据处理层：解析、聚合、统计，生成结构化数据
• 可视化展示层：用终端表格、HTML报告、或者简单的Web页面展示

我个人习惯用Python写这类工具，因为标准库丰富，解析procfs和sysfs非常方便。当然你也可以用Shell + awk，但可维护性差一些。

30.2 核心代码实现

我们直接上代码。先写一个Wakeup源监控类，抓取/sys/kernel/debug/wakeup_sources的数据。

import os
import time
import json
from datetime import datetime

class WakeupSourceMonitor:
    """Wakeup源监控器"""
    
    def __init__(self, wakeup_sources_path="/sys/kernel/debug/wakeup_sources"):
        self.path = wakeup_sources_path
        self.history = []
        
    def read_raw(self):
        """读取原始数据"""
        if not os.path.exists(self.path):
            raise FileNotFoundError(f"找不到 {self.path}，请确认内核已开启CONFIG_DEBUG_FS")
        with open(self.path, 'r') as f:
            return f.read()
    
    def parse(self, raw_text):
        """解析为结构化数据"""
        lines = raw_text.strip().split('\n')
        if len(lines) < 2:
            return []
        # 第一行是表头，跳过
        header = lines[0].split()
        sources = []
        for line in lines[1:]:
            parts = line.split()
            if len(parts) < 6:
                continue
            source = {
                'name': parts[0],
                'active_count': int(parts[1]),
                'event_count': int(parts[2]),
                'wakeup_count': int(parts[3]),
                'expire_count': int(parts[4]),
                'active_since': int(parts[5]),
                'total_time': int(parts[6]) if len(parts) > 6 else 0,
                'max_time': int(parts[7]) if len(parts) > 7 else 0,
                'last_change': int(parts[8]) if len(parts) > 8 else 0,
                'prevent_suspend_time': int(parts[9]) if len(parts) > 9 else 0,
            }
            sources.append(source)
        return sources
    
    def snapshot(self):
        """采集一次快照"""
        raw = self.read_raw()
        data = self.parse(raw)
        record = {
            'timestamp': datetime.now().isoformat(),
            'sources': data
        }
        self.history.append(record)
        return record
    
    def get_top_wakeups(self, n=5):
        """获取唤醒次数最多的前N个源"""
        if not self.history:
            return []
        latest = self.history[-1]['sources']
        sorted_sources = sorted(latest, key=lambda x: x['wakeup_count'], reverse=True)
        return sorted_sources[:n]
    
    def detect_abnormal(self, threshold_wakeups=100):
        """检测异常Wakeup源（唤醒次数超过阈值）"""
        if not self.history:
            return []
        latest = self.history[-1]['sources']
        abnormal = [s for s in latest if s['wakeup_count'] > threshold_wakeups]
        return abnormal

30.3 唤醒锁监控模块

Wakeup源看完了，还得看唤醒锁。我们用dumpsys power来抓。

import subprocess
import re

class WakeLockMonitor:
    """唤醒锁监控器"""
    
    def __init__(self):
        self.history = []
        
    def get_wakelocks(self):
        """通过dumpsys获取当前唤醒锁"""
        try:
            output = subprocess.check_output(
                ['adb', 'shell', 'dumpsys', 'power'],
                timeout=5
            ).decode('utf-8', errors='ignore')
        except:
            # 如果没有adb，尝试本地执行（需要root）
            try:
                output = subprocess.check_output(
                    ['dumpsys', 'power'],
                    timeout=5
                ).decode('utf-8', errors='ignore')
            except:
                return []
        
        return self._parse_wakelocks(output)
    
    def _parse_wakelocks(self, text):
        """解析唤醒锁信息"""
        wakelocks = []
        # 匹配 "Wake Locks: size=xxx" 后面的内容
        in_wakelock_section = False
        for line in text.split('\n'):
            if 'Wake Locks:' in line:
                in_wakelock_section = True
                continue
            if in_wakelock_section:
                if line.strip() == '':
                    break
                # 格式: "  lock_name (type)  held_count=xxx  ..."
                match = re.match(r'\s*(.+?)\s+\((\w+)\)\s+.*held_count=(\d+)', line)
                if match:
                    wakelocks.append({
                        'name': match.group(1),
                        'type': match.group(2),
                        'held_count': int(match.group(3)),
                    })
        return wakelocks
    
    def snapshot(self):
        """采集一次快照"""
        wl = self.get_wakelocks()
        record = {
            'timestamp': datetime.now().isoformat(),
            'wakelocks': wl
        }
        self.history.append(record)
        return record
    
    def get_held_wakelocks(self):
        """获取当前正在持有的唤醒锁"""
        latest = self.snapshot()
        # 通常held_count > 0表示正在持有
        held = [wl for wl in latest['wakelocks'] if wl['held_count'] > 0]
        return held

30.4 可视化展示

数据有了，怎么展示？我推荐两种方式：

1. 终端表格：用prettytable库，适合快速调试
2. HTML报告：生成带图表的报告，适合归档和分享

这里给出HTML报告生成的核心代码：

def generate_html_report(wakeup_data, wakelock_data, output_path="power_report.html"):
    """生成电源管理HTML报告"""
    html = []
    html.append('<!DOCTYPE html>')
    html.append('<html><head><meta charset="utf-8">')
    html.append('<title>电源管理监控报告</title>')
    html.append('<style>')
    html.append('body{font-family:Arial,sans-serif;margin:20px;background:#f5f5f5}')
    html.append('h1{color:#2c3e50}')
    html.append('table{border-collapse:collapse;width:100%;margin:16px 0}')
    html.append('th,td{border:1px solid #ddd;padding:8px;text-align:left}')
    html.append('th{background:#4a90d9;color:white}')
    html.append('tr:nth-child(even){background:#f2f2f2}')
    html.append('.abnormal{background:#ffebee;color:#c62828}')
    html.append('</style></head><body>')
    
    html.append(f'<h1>电源管理监控报告</h1>')
    html.append(f'<p>生成时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</p>')
    
    # Wakeup源表格
    html.append('<h2>Wakeup源统计</h2>')
    html.append('<table><tr><th>名称</th><th>唤醒次数</th><th>活跃时间(ms)</th><th>阻止挂起时间(ms)</th></tr>')
    for src in wakeup_data:
        cls = 'abnormal' if src['wakeup_count'] > 100 else ''
        html.append(f'<tr class="{cls}">')
        html.append(f'<td>{src["name"]}</td>')
        html.append(f'<td>{src["wakeup_count"]}</td>')
        html.append(f'<td>{src["total_time"]}</td>')
        html.append(f'<td>{src["prevent_suspend_time"]}</td>')
        html.append('</tr>')
    html.append('</table>')
    
    # 唤醒锁表格
    html.append('<h2>当前唤醒锁</h2>')
    html.append('<table><tr><th>名称</th><th>类型</th><th>持有计数</th></tr>')
    for wl in wakelock_data:
        html.append(f'<tr><td>{wl["name"]}</td><td>{wl["type"]}</td><td>{wl["held_count"]}</td></tr>')
    html.append('</table>')
    
    html.append('</body></html>')
    
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write('\n'.join(html))
    print(f"报告已生成: {output_path}")

30.5 自动化测试脚本

工具写好了，不能每次都手动跑。我们要把它集成到CI/CD里。下面是一个自动化测试脚本的框架：

#!/usr/bin/env python3
"""
电源管理自动化测试脚本
用于持续集成环境，自动采集数据并生成报告
"""

import sys
import time
import argparse
from pathlib import Path

# 导入我们上面写的模块
# from wakeup_monitor import WakeupSourceMonitor
# from wakelock_monitor import WakeLockMonitor
# from report_generator import generate_html_report

def run_test(duration=60, interval=5, output_dir="./reports"):
    """运行电源管理测试"""
    print(f"[INFO] 开始电源管理测试，持续{duration}秒，每{interval}秒采样一次")
    
    # 初始化监控器
    # wsm = WakeupSourceMonitor()
    # wlm = WakeLockMonitor()
    
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    
    all_wakeup_data = []
    all_wakelock_data = []
    
    start_time = time.time()
    sample_count = 0
    
    while time.time() - start_time < duration:
        sample_count += 1
        print(f"[INFO] 第{sample_count}次采样...")
        
        # 采集数据
        # wakeup_record = wsm.snapshot()
        # wakelock_record = wlm.snapshot()
        
        # 检测异常
        # abnormal = wsm.detect_abnormal(threshold_wakeups=50)
        # if abnormal:
        #     print(f"[WARN] 发现异常Wakeup源: {[s['name'] for s in abnormal]}")
        
        # 收集数据用于最终报告
        # all_wakeup_data.extend(wakeup_record['sources'])
        # all_wakelock_data.extend(wakelock_record['wakelocks'])
        
        time.sleep(interval)
    
    # 生成报告
    report_path = output_path / f"power_report_{int(time.time())}.html"
    # generate_html_report(all_wakeup_data, all_wakelock_data, str(report_path))
    
    print(f"[INFO] 测试完成，共采样{sample_count}次")
    print(f"[INFO] 报告已保存到: {report_path}")
    
    # 返回测试结果（用于CI判断）
    # if len(abnormal) > 0:
    #     return 1  # 失败
    return 0  # 成功

def main():
    parser = argparse.ArgumentParser(description='电源管理自动化测试')
    parser.add_argument('--duration', type=int, default=60, help='测试持续时间(秒)')
    parser.add_argument('--interval', type=int, default=5, help='采样间隔(秒)')
    parser.add_argument('--output', type=str, default='./reports', help='报告输出目录')
    args = parser.parse_args()
    
    exit_code = run_test(args.duration, args.interval, args.output)
    sys.exit(exit_code)

if __name__ == '__main__':
    main()

30.6 实战经验总结

最后，分享几个我在实际项目中踩过的坑：

• 权限问题：/sys/kernel/debug/wakeup_sources需要root权限。我曾经在user版本上折腾了半天，最后发现adb root才能读。建议工具里加一个权限检测，提前提示用户。
• 数据过时：wakeup_sources里的计数器是累计值，不是瞬时值。要计算“增量”才能知道当前时间段内的唤醒频率。我一开始没注意，直接拿绝对值做分析，得出了完全错误的结论。
• 唤醒锁泄漏：有些App会长时间持有PARTIAL_WAKE_LOCK不放。我建议在监控工具里加一个“持有时间超过30秒”的告警，这个阈值可以根据产品需求调整。
• 多设备支持：不同厂商的内核版本可能修改了wakeup_sources的输出格式。我在某款MTK平台上就遇到过字段顺序不一样的情况。建议解析时用字段名匹配，而不是按位置取。

好了，整个课程到这里就结束了。从第一节课的电源管理基础概念，到今天的综合实战，我们走完了Android电源管理的完整知识体系。希望这些内容能帮你在实际工作中少走弯路。

记住，电源管理没有银弹。每个问题都要结合具体场景分析。但有了这套工具和方法论，至少你不会再像无头苍蝇一样乱撞了。

---

公众号:蓝海资料掘金营，微信deep3321