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.保存到本地的数据不会从中间截取某段时间了。
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 # 导出文件所保存的类型，
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
