DirkZhao
8/19/2019 - 12:43 PM

线程使用情况测试

import re
import time

from pandas import DataFrame

from AdbTest.Utils.CmdUtils import shell


def look_for_pid():
    sto = shell('adb shell ps -ef | grep com.jifen.qukan')
    if len(sto) > 20:
        pid = sto.split()[1]
        pid_str = str(pid)
        print('这次的pid是:' + pid_str)
        return pid
    else:
        print('adb命令出错')


def threads_statistics(theme, model=1):
    """
    根据model来输出多种数据:
    1. 原始数据
    2. GroupBy之后,按照数量排序。
    """
    # 根据pid查找对应的子线程
    cmd = 'adb shell ps -l ' + look_for_pid() + ' -T'
    result = shell(cmd)
    # print(result)
    # 正则,通过换行符切割
    items = re.split('\\\\n', result)
    # 去掉无用数据
    items = items[1:-1]

    data = {'线程名': [], '线程id': []}
    for item in items:
        it = re.split(' ', item)
        tid = it[4]
        name = it[-1]
        data['线程名'].append(name)
        data['线程id'].append(tid)
    df = DataFrame(data)
    # 输出到对应文件
    now = time.strftime('%H:%M:%S ', time.localtime(time.time()))
    file = '/Users/rockyzhao/Desktop/' + theme + '_' + now + '.csv'
    if model == 1:
        df.to_csv(file)
        pass
    if model == 2:
        # 根据线程名count
        df = df.groupby(by='线程名').count()
        # 根据线程id排序
        df = df.sort_values('线程id', ascending=False)
        df.to_csv(file)
        pass
    else:
        print('model值不对')
        pass


threads_statistics(theme='cpcswitch', model=2)