（图 1-1）
目前公司性能压测大多是对单接口的性能压测，这种压测策略固然能很好的反映被测系统接口的性能，能确保每一个接口的性能都处在合理范围内，但跟实际情况还是有点出入。在实际的生产中，某个服务通常会提供多个接口，在对外提供服务时，各个接口存在一定比例结构混合调用，通常我们所说的业务模型就是指这个。后面会对这个业务模型进行详细的说明。
业务模型调研是性能测试前非常重要的一个调研环节。拥有一个完整的业务模型，可以最真实的还原出生产的业务情况，只有模型符合实际的生产业务操作，性能测试的结果才能真实有效的反映上线生产系统的实际性能情况。
我们可以通过分析生产环境的历史数据来确定各种业务量数据在整个系统压力所占比例，最终目的是建立生产应用系统实际运行的业务模型。而中通自研的 ZCAT 平台为测试人员获取生产数据提供了源头，通过 ZCAT 的接口可以获取被测应用的调用数（分钟维度）。
该业务模型实现原理是：基于 python 实现业务模型的自动构建，辅助以人为的分析，判定出当前服务的生产实际压测模型。测试人员可以通过这个模型来实现对压测流量的分配。人为分析部分主要依赖 Python 办公自动化功能生成的各个阶段的图表，涉及的主要 python 库有 openpyxl（处理 excel），requests（接口调用）。
接下来详细看下这个业务模型整理的过程：
（1）调用 ZCAT 接口获取最近 15 天数据（ZCAT 目前只支持最近 7 天的数据查询，后续使用 7 天的数据量进行分析），通过数据统计与计算，输出最近 7 天的被测服务下所有业务的调用量分布图（图 1-1）：
（图 1-1）
通过（图 1-1）我们可以看出，最近 7 天的业务水平比较接近，业务峰值在 5-18（业务量是 386,000,000）。同时，基于上面的数据计算每一天所有业务占总调用量的百分比。如下图（图 1-2）可见，在每一天中，各项业务量的占比基本一致，这是一个比较典型的服务，整体水平比较平稳。
（图 1-2）
当我们选择出需要分析的时间点后，我们需要针对这一天的业务进行细致分析，输出每小时（5 月 18 号）的业务调用情况。通过下图（图 1-3）可知，业务高峰出现在 00 点，但是值得注意的是 23 点，这个时间点的业务量也比较大，而且这个小时的业务比例与 00 点完全不同，当处在这样的业务场景下，服务本身对机器的性能消耗是完全不一样的，所以这个节点也可以作为被分析对象。
（图 1-3）
通过上述的数据计算每小时内每个服务占总调用量的百分比，根据下图（图 1-4）展示的比例可以基本得出我们需要的 TPS 比例：
（图 1-4）
（2） TPS 计算：
a.通用模型：
i.将这一天的所有业务数加在一起，再将各业务整天的交易量加在一起，计算各业务量的比例。
ii.通过这个模型我们能计算出一个通用的 TPS：
1.TPS = 386000000/(24*3600) = 4467。也就是说在通用场景中，TPS 不能低于 4467
b.5 月 18 号，23 点 TPS：
i.通过 23 点的业务量我们计算：
1.TPS = 28885649/3600 = 8023。也就是说在 23 点场景中，TPS 不能低于 8023
c.5 月 18 号，00 点 TPS：
i.通过 00 点的业务量我们计算：
1.TPS = 31103971/3600 = 8639。也就是说在 00 点场景中，TPS 不能低于 8639
（3）业务模型设计：
a. 像 23 和 00 点这种，虽然 TPS 差不多，但是业务比例差别大，这两种业务模型下，对系统资源的消耗会完全不一样，所以我们设计整个压测模型时，需要将这两种常见囊括在内。根据下图（图 1-5）汇总的数据进行业务模型 TPS 的分配如下。
b. 23 点业务模型：
i. TPS > 8639
ii. 业务分配比例：
业务 4 -> 1%
业务 6 -> 5%
业务 7 -> 1%
业务 8 -> 93%
c. 00 点业务模型：
i. TPS > 8023
ii. TPS 分配比例：
业务 6 -> 43%
业务 7 -> 1%
业务 8 -> 56%
（图 1-5）
通过多个维度的图表数据进行分析总结，我们可以得到比较符合生产场景的业务 TPS 分布，压测就能有针对性的进行业务配比，使得我们的压测更有效地反映整体服务的性能瓶颈。
下面附上关键节点的源码，有需要的同学欢迎讨论。
源码：
#!/usr/bin/env python
# encoding: utf-8
import os
import time
from copy import deepcopy
from datetime import datetime,date,timedelta
import requests
from openpyxl import Workbook, load_workbook
from openpyxl.chart import (
LineChart,
Reference, BarChart, label,
)
'''
性能测试-业务模型构建
根据ZCat上爬取的数据生成图表
'''
# 基础数据
app_id = '{appid}'
file_location = '{file_location}'
user_agent = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.114 Safari/537.36'
cookie = 'cookie'
col_name = ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U',
'V', 'W', 'X', 'Y', 'Z'
, 'AA', 'AB', 'AC', 'AD', 'AE', 'AF', 'AG', 'AH', 'AI', 'AJ', 'AK', 'AL', 'AM', 'AN', 'AO', 'AP', 'AQ', 'AR',
'AS', 'AT', 'AU', 'AV', 'AW', 'AX', 'AY', 'AZ'
, 'BA', 'BB', 'BC', 'BD', 'BE', 'BF', 'BG', 'BH', 'BI', 'BJ', 'BK', 'BL']
# 保存数据到excel，并生成折线图
def save_data_to_excel( file_name, target_sheet_name,data_dic,title_list):
if not os.path.exists(file_name):
wb = Workbook()
else:
wb = load_workbook(file_location)
sheet_name_list = wb.sheetnames
if target_sheet_name in sheet_name_list:
wb.remove(wb[target_sheet_name])
sheet = wb.create_sheet(target_sheet_name)
# 输入表头
col_index=1
for each_date in title_list:
sheet['%s1' % col_name[col_index]] = each_date
col_index +=1
# 输入行标题
start_name_index = 2
for each_name in data_dic:
sheet['A%d' % start_name_index] = each_name
start_name_index += 1
# 输入数据
start_name_index = 2
for key,value in data_dic.items():
col_index = 1
for v in value:
sheet['%s%d' % (col_name[col_index], start_name_index)] = v
col_index += 1
start_name_index += 1
# 获得最大列号
col_max=1
for key, value in data_dic.items():
if len(value) > col_max:
col_max=len(value)
sheet.column_dimensions['A'].width = 20
for service in data_dic.items():
sheet.column_dimensions[col_name[col_index]].width = 13
# 绘制折线图
add_line_chart(target_sheet_name, sheet, 1, 2, col_max+1, len(data_dic.keys()) + 1)
# 保存文件
wb.save(file_name)
wb.close()
# 保存数据到excel，并生成柱状图
def save_data_to_excel_bar(file_name,title,target_sheet_name,data_dic):
if not os.path.exists(file_name):
wb = Workbook()
else:
wb = load_workbook(file_location)
sheet_name_list = wb.sheetnames
if target_sheet_name in sheet_name_list:
wb.remove(wb[target_sheet_name])
sheet = wb.create_sheet(target_sheet_name)
rows = []
for v in data_dic.values():
rows.append(tuple(v))
# 插入表头
index = 1
for v in data_dic.keys():
sheet['%s%d' % (col_name[index], 1)] = v
index += 1
# 插入行标题
index2 = 2
for v in ['{num:02d}'.format(num=i) for i in range(24)]:
sheet['%s%d' % (col_name[0], index2)] = v
index2 += 1
#插入数据
start_name_index = 1
for row in rows:
col_index = 2
for v in row:
sheet['%s%d' % (col_name[start_name_index],col_index)] = v
col_index += 1
start_name_index += 1
# 插入饼状图
if title == 'Daily_per_hour':
is_show_value = False
else:
is_show_value = True
add_bar_chart(title,sheet,len(data_dic.keys()),is_show_value)
# 保存文件
wb.save(file_name)
wb.close()
def save_data_to_excel_sub( file_name, target_sheet_name,data_dic,target_date,target_hour):
if not os.path.exists(file_name):
wb = Workbook()
else:
wb = load_workbook(file_location)
sheet_name_list = wb.sheetnames
if target_sheet_name in sheet_name_list:
wb.remove(wb[target_sheet_name])
sheet = wb.create_sheet(target_sheet_name)
# Excel中插入数据部分源码
# 输入表头
sheet['B1'] = target_date+' 业务百分比'
tmp = 2
for hour in target_hour:
sheet['%s%d' % (col_name[tmp], 1)] = '第' + hour + '小时的业务百分比'
tmp += 1
# 输入行标题
start_name_index = 2
for each_name in data_dic.keys():
sheet['A%d' % start_name_index] = each_name
start_name_index += 1
# 输入数据
start_name_index = 2
for value in list(data_dic.values()):
col_index = 1
for v in value:
sheet['%s%d' % (col_name[col_index], start_name_index)] = round(v,2)
col_index += 1
start_name_index += 1
# 保存文件
wb.save(file_name)
wb.close()
# 生成折线图
def add_line_chart(title, wss, min_col, min_row, max_col, max_row):
c1 = LineChart() # 生成饼图BarChart()
c1.width = 50
c1.height = 15
c1.title = title # 图的标题
c1.style = 12 # 线条的style
c1.y_axis.title = 'count' # y坐标的标题
c1.x_axis.title = "Date" # x坐标的标题
data = Reference(wss, min_col=min_col, min_row=min_row, max_col=max_col, max_row=max_row) # 图像的数据 起始行、起始列、终止行、终止列
c1.add_data(data, titles_from_data=True, from_rows=True)
dates = Reference(wss, min_col=2, min_row=1, max_col=max_col)
c1.set_categories(dates)
wss.add_chart(c1, "B12") # 将图表添加到 sheet中
# 生成柱状图
def add_bar_chart(title, wss,max_col,is_show_value=False):
chart1 = BarChart()
chart1.type = "col"
chart1.style = 10
chart1.title = title
chart1.y_axis.title = 'Test number'
chart1.x_axis.title = 'Sample length (mm)'
data = Reference(wss, min_col=2, min_row=1, max_row=25, max_col=max_col+1)
cats = Reference(wss, min_col=1, min_row=2, max_row=26)
chart1.add_data(data, titles_from_data=True)
chart1.set_categories(cats)
chart1.shape = 4
chart3 = deepcopy(chart1)
chart3.type = "col"
chart3.style = 2
chart3.height = 26 # default is 7.5
chart3.width = 45 # default is 15
chart3.grouping = "stacked"
chart3.overlap = 100
chart3.y_axis.title = "业务量"
chart3.x_axis.title = "每小时时序轴"
chart3.title = title
if is_show_value:
chart3.dLbls = label.DataLabelList()
chart3.dLbls.showVal=True
wss.add_chart(chart3, "A27")
# 获取当前日期前7天的日期，返回日期列表
def get_7_days_before_now():
days_before = []
the_date = date.today()
for i in range(1,8).__reversed__():
days_before.append((the_date + timedelta(days=-i)).strftime('%Y-%m-%d'))
return days_before
def format_date_to_timestamp(date):
timestamp = time.mktime(time.strptime(date, "%Y-%m-%d %H:%M:%S"))*1000
real_time = str(timestamp).split('.')[0]
return real_time
#根据appid调用ZCAT接口，获取所有服务最近7天的调用量，以天为单位
def least_7_days_data_per_day(app_id):
result = {}
inner_total = {}
inner_metric_details = {}
for day in get_7_days_before_now():
start_at = format_date_to_timestamp(day + ' 00:00:00')
end_at = format_date_to_timestamp(day + ' 23:59:59')
url = '{backend_service_api}'
response = requests.get(url).json()
for detail in response:
if detail['name'] == 'rpc.dubbo.service':
inner_metric_details[day] = detail['metricDetails']
if detail['name'] == '$$TOTAL':
for d in detail['metricDetails']:
if d['name'] == 'rpc.dubbo.service':
inner_total[day] = d['count']
result['metricDetails'] = inner_metric_details
result['total'] = inner_total
percent_dic = {}
service_daily_dic = {}
for date,metrics in inner_metric_details.items():
total = inner_total[date]
for m in metrics :
if m['name'] not in percent_dic.keys():
percent_dic[m['name']] = []
service_daily_dic[m['name']] = []
percent_dic[m['name']].append(m['count'] / total)
service_daily_dic[m['name']].append(m['count'])
result['percentDetails'] = percent_dic
result['dailyCount'] = service_daily_dic
return result
# 当回指定日期当天的业务分布数据，以每小时为单位进行聚合
def today_data_per_hour_count(day,service_name):
start_at = format_date_to_timestamp(day+' 00:00:00')
end_at = format_date_to_timestamp(day+' 23:59:59')
hour_list = ['{num:02d}'.format(num=i) for i in range(24)]
value = [0] * 24
result = {}
for name in service_name:
url = '{backend_service_api}'
response = requests.get(url).json()
inner_result = dict(zip(hour_list,value))
for detail in response.get('hitsOverTime'):
hour = datetime.fromtimestamp(int(detail['time'])/1000).strftime('%H')
inner_result[hour] = inner_result.get(hour) + detail['value']
result[name] = inner_result.values()
return result
# 返回指定日期当天的业务分布的百分比分布，以每小时为单位
def today_data_per_hour_percent(data):
hours = ['{num:02d}'.format(num=i) for i in range(24)]
per_hour_count = {}
index = 0
for h in hours:
count = 0
for row in data.values():
count = count + (list(row))[index]
index += 1
per_hour_count[h] = count
for k,v in data.items():
index2 = 0
new_list = list(v)
for h in hours:
if per_hour_count[h] == 0:
new_list[index2] = 0
else:
new_list[index2] = round(new_list[index2]/per_hour_count[h], 2)
index2 += 1
data[k] = new_list
return data
# 获取目标当天的业务百分比分布数据
def today_data_percent(data,day,hours,service_name):
today_total_count = data['total'][day]
result = {}
for detail in data['metricDetails'][day]:
result[detail['name']] = [round(detail['count']/today_total_count,2)]
per_hour_count = today_data_per_hour_count(day, service_name)
total_count = [0] * 24
for detail in per_hour_count.values():
index = 0
for d in list(detail):
total_count[index] = total_count[index] + d
index += 1
for hour in hours:
for k, v in per_hour_count.items():
result[k].append(list(v)[int(hour)] / total_count[int(hour)])
return result
if __name__ == '__main__':
# 获取每一天的总调用量
least_7_days_all_data = least_7_days_data_per_day(app_id)
# 生成'最近15天所有服务的总量分布图，以天为单位'
date_list = list(least_7_days_all_data['metricDetails'].keys())
save_data_to_excel(file_location, '近7天业务量分布', least_7_days_all_data['dailyCount'], date_list)
# 生成'最近15天所有服务的总量分布图，以百分比为单位'
save_data_to_excel(file_location, '近7天业务百分比分布', least_7_days_all_data['percentDetails'], date_list)
# 生成'业务峰值当天的服务调用分布柱状图'
service_list = least_7_days_all_data['percentDetails'].keys()
target_date = '2021-05-18'
today_data_static = today_data_per_hour_count(target_date,service_list)
save_data_to_excel_bar(file_location, '业务峰值当天的服务调用分布柱状图','每小时业务分布', today_data_static)
# 生成'业务峰值当天的服务调用百分比分布柱状图'
save_data_to_excel_bar(file_location, '业务峰值当天的服务调用百分比分布柱状图','每小时业务百分比分布',
today_data_per_hour_percent(today_data_static))
# 根据用户输入，展示业务分布图表
hour_list = ['00','23']
# 2021-05-17这一天的业务百分比分布
today_data_percent_static = today_data_percent(least_7_days_all_data,target_date,hour_list,service_list)
# 写入excel表格
save_data_to_excel_sub(file_location,'业务分布图',today_data_percent_static,target_date,hour_list)
