从零开始玩转 PerfDogService Demo解析篇
1.Demo文件夹
当下载好PerfDogService后文件夹是这个样子的。
文件说明：
PerfDogService.exe： Service启动文件，一般不用手工启动，自动化脚本会自动启动此程序。
perfdog.proto: grpc接口和protobuf结构体的定义文件
demo/python: python3的demo,依赖库grpcio和protobuf
demo/java: java的demo,依赖库grpc-netty-shaded, grpc-protobuf, grpc-stub
PerfDogService.jar: 服务所依赖的jar包
我们先不管其他，只专注于demo文件夹；
展开后是Java和Python两种语言的Demo，我们本次只分析改造Python的
Python的Demo只有三个文件
Demo.py: 我们程序的执行入口，同时我们主要操作的也是这个文件；
perfdog_pb2.py： 此文件是由协议缓冲区编译器生成，也就是由perfdog.proto这个文件生成的。我们不要修改它。
perfdog_pb2_grpc.py： 此文件是gRPC Python协议编译器插件生成，也不要修改它
这里简单介绍下==协议缓冲区==
协议缓冲区是是一种与语言无关，平台无关的数据序列化结构化机制；
只需要一次构造数据，然后就可以使用生成的特殊源代码轻松地使用各种语言在各种数据流中写入和读取结构化数据；
你用python语言，你辛辛苦苦用python定义好了一堆字典啦或者Class啦或者其他什么数据，你想给隔壁老李用，然而老李用java啊，这时候隔壁小王也需要用，然而小王用C++啊，各种语言不能直接调用。这个时候通过协议缓冲区这种机制，先把你的数据编译生成一种特殊的源代码，老李和小王就可以调用这份源代码了。
2.Demo.py文件解析
其实全篇只有一个Run函数，而且官方写的注释也比较清晰了，我们就来简单看下吧
0到3其实没什么好说的了，就是唤起PerfDogService服务，通过令牌和服务器建立了一个连接，其中stub除了建立连接还有好多其他的方法
接下来看看第四步
deviceEventIterator这里是一个可迭代对象，存放着当前所有的连接设备，例如Android啦，Ios，模拟器啦等等；
我们最常用的应该是device，这里保存着被测机型的uid，名字等。这里的uid就是当前连接设备的唯一标识，如果你安装了ADB，可以在控制台输入ADB devices查看你当期连接的设备信息，例如这样：
5其实就是开始初始化设备
6是开始获取当前被测设备的所有安装的应用，应用保存在apps这个对象里，这也是一个可迭代对象，
迭代apps可以获取到每个app，以及每个app的名字，包名，版本，Icon，是不是系统App，subVersion等等。
第7步这里实际上是获取了当前被测设备的硬件信息，例如像CPU，型号，安卓版本之类的等等
第8步这里开始选择要搜集的性能数据指标都有哪些，我这里多加了一个CPU Core的使用率
第9步就开始搜集性能数据了
采集性能数据信息实际上通过多线程的方式进行的，所以Sleep(20)就是采集20s的数据
10和11就是添加不同时间段的注释
第12步是两个操作
1.上传整个数据到云端
2.导出第5秒到20秒的数据到本地，其中导出路径可以自己更改（outputDirectory）
13最后停止测试，不过其实进程没停止，如果此时你启动Perfdog客户端，可能会显示这个错误。
此时去Kill掉这个java进城就可以了。
3.Demo简单修改
由于我个人的需求1是不想每次配置都挨行代码修改，2是为后续搭建自定义的云性能测试平台做铺垫，所以简单修改了下代码：源代码未删，仅是注释掉了，方便做对比；
1.提取常用配置
2.需要提前配置好需要测试的对应设备
3.需要提前配置好要测试的APP（配置APP的名字）
4.只上传一份数据到云端，
5.保存到本地的数据不会从中间截取某段时间了。
# -*- coding: utf-8 -*-
import subprocess
import time
import traceback
import grpc
import perfdog_pb2_grpc
import perfdog_pb2
import threading
class Config():
PERFDOGSERVER_PATH=r'C:\Users\Administrator\Desktop\perfdog\PerfDogService.exe' #PerfDogService的路径
TOKEN='XXXXXXXXXXXXXXXXXXXX '#token
DEVICES_UID="813QEDTE228ZK" #需要测试的手机序列号
APP_NAME="天气" #需要测试的app名字
TEST_TIMER=11 #需要测试时长
LABEL="this is a label" #label内容
NOTE="this is a note" #note内容
CASENAME="天气测试数据" #用例描述 ,会覆盖相同用例名字的数据
SAVEDATA_BEGINTIME=1 #导出到本地数据的开始时间(秒)
SAVEDATA_ENDTIME=20 #导出到本地数据的结束时间（秒）
OUTPUT="F:\\perfdog_service_output\\" #导出文件所保存的目录，
SAVEDATA_TYPE=perfdog_pb2.EXPORT_TO_JSON #导出文件所保存的类型，
# 第一次运行demo前需要通过pip安装grpcio(1.23.0)和protobuf(3.10.0)
def run():
try:
# 在代码里启动PerfDogService或手动启动PerfDogService
print("0.启动PerfDogService")
# 1.**************************填入PerfDogService的路径
perfDogService = subprocess.Popen(Config.PERFDOGSERVER_PATH)
# 等待PerfDogService启动完毕
time.sleep(5)
print("1.通过ip和端口连接到PerfDog Service")
options = [('grpc.max_receive_message_length', 100 * 1024 * 1024)]
channel = grpc.insecure_channel('127.0.0.1:23456', options=options)
print("2.新建一个stub,通过这个stub对象可以调用所有服务器提供的接口")
stub = perfdog_pb2_grpc.PerfDogServiceStub(channel)
print("3.通过令牌登录，令牌可以在官网申请")
userInfo = stub.loginWithToken(perfdog_pb2.Token(token=Config.TOKEN))
print("UserInfo:\n", userInfo)
print("4.启动设备监听器监听设备,每当设备插入和移除时会收到一个DeviceEvent")
deviceEventIterator = stub.startDeviceMonitor(perfdog_pb2.Empty())
for deviceEvent in deviceEventIterator:
# 从DeviceEvent中获取到device对象，device对象会在后面的接口中用到
device = deviceEvent.device
print("当前devices： ",device," **** ",deviceEvent)
# time.sleep(20000)
if deviceEvent.eventType == perfdog_pb2.ADD and device.uid==Config.DEVICES_UID:
print("设备[%s:%s]插入\n" % (device.uid, perfdog_pb2.DEVICE_CONTYPE.Name(device.conType)))
# 每台手机会返回两个conType不同的设备对象(USB的和WIFI的),如果是测有线，取其中的USB对象
if device.conType == perfdog_pb2.USB:
print("5.初始化设备[%s:%s]\n" % (device.uid, perfdog_pb2.DEVICE_CONTYPE.Name(device.conType)))
stub.initDevice(device)
print("6.获取app列表")
appList = stub.getAppList(device)
#
apps = appList.app
app_index = 0
for app in apps:
print('%s: %s->%s' % (app_index, app.label, app.packageName))
if app.label==Config.APP_NAME:
app_select=app_index
break
else:
app_index += 1
app_select=None
if app_select is None:app_select = int(input("未安装输入APP，请选择要测试App: "))
app = apps[app_select]
print("7.获取设备的详细信息")
deviceInfo = stub.getDeviceInfo(device)
print("8.开启性能数据项")
stub.enablePerfDataType(
perfdog_pb2.EnablePerfDataTypeReq(device=device, type=perfdog_pb2.NETWORK_USAGE))
stub.enablePerfDataType(
perfdog_pb2.EnablePerfDataTypeReq(device=device, type=perfdog_pb2.NORMALIZED_CPU_CORE_USAGE))
print("9.开始收集[%s:%s]的性能数据\n" % (app.label, app.packageName))
print(stub.startTestApp(perfdog_pb2.StartTestAppReq(device=device, app=app)))
req = perfdog_pb2.OpenPerfDataStreamReq(device=device)
perfDataIterator = stub.openPerfDataStream(req)
def perf_data_process():
for perfData in perfDataIterator:
print(perfData)
threading.Thread(target=perf_data_process).start()
# 采集一些数据
time.sleep(Config.TEST_TIMER)
print("10.设置label")
stub.setLabel(perfdog_pb2.SetLabelReq(device=device, label="I am a label"))
time.sleep(3)
print("11.添加批注")
stub.addNote(perfdog_pb2.AddNoteReq(device=device, time=5000, note="I am a note"))
print("12.上传和导出所有数据")
saveResult = stub.saveData(perfdog_pb2.SaveDataReq(
device=device,
caseName=Config.CASENAME, # web上case和excel的名字
uploadToServer=True, # 上传到perfdog服务器
exportToFile=True, # 保存到本地
outputDirectory=Config.OUTPUT,
dataExportFormat=Config.SAVEDATA_TYPE
))
print("保存结果:\n", saveResult)
# print("12.上传和导出第5秒到20秒的数据")
# stub.saveData(perfdog_pb2.SaveDataReq(
# device=device,
# beginTime=5000, # 指定开始时间
# endTime=20000, # 指定结束时间
# caseName="case2", # web上case和excel的名字
# uploadToServer=True, # 上传到perfdog服务器
# exportToFile=True, # 保存到本地
# outputDirectory="F:\\perfdog_service_output\\",
# dataExportFormat=perfdog_pb2.EXPORT_TO_EXCEL
# ))
print("13.停止测试")
stub.stopTest(perfdog_pb2.StopTestReq(device=device))
print("over")
break
elif deviceEvent.eventType == perfdog_pb2.REMOVE:
print("设备[%s:%s]移除\n" % (device.uid, perfdog_pb2.DEVICE_CONTYPE.Name(device.conType)))
except Exception as e:
traceback.print_exc()
if __name__ == '__main__':
run()
#
性能测试技术交流群：720150565
查看PerfDog详情：https://perfdog.qq.com/?ADTAG=media.dev_website