257 lines
8.0 KiB
Go
257 lines
8.0 KiB
Go
// Package statistics 统计数据
|
||
package statistics
|
||
|
||
import (
|
||
"fmt"
|
||
"io/ioutil"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"go_dreamfactory/lego/core"
|
||
"go_dreamfactory/lego/sys/log"
|
||
"go_dreamfactory/stress/tools"
|
||
|
||
"go_dreamfactory/stress/model"
|
||
|
||
"golang.org/x/text/language"
|
||
"golang.org/x/text/message"
|
||
"gopkg.in/yaml.v2"
|
||
)
|
||
|
||
var (
|
||
// 输出统计数据的时间
|
||
exportStatisticsTime = 1 * time.Second
|
||
p = message.NewPrinter(language.English)
|
||
requestTimeList []uint64 // 所有请求响应时间
|
||
)
|
||
|
||
type Options struct {
|
||
ConfPath string //配置文件路径
|
||
Version string //服务版本
|
||
Setting core.ServiceSttings //服务参数配置
|
||
}
|
||
|
||
func newOptions(path string) *Options {
|
||
options := &Options{}
|
||
|
||
yamlFile, err := ioutil.ReadFile(path)
|
||
if err != nil {
|
||
panic(fmt.Sprintf("读取服务配置【%s】文件失败err:%v:", options.ConfPath, err))
|
||
}
|
||
err = yaml.Unmarshal(yamlFile, &options.Setting)
|
||
if err != nil {
|
||
panic(fmt.Sprintf("读取服务配置【%s】文件失败err:%v:", options.ConfPath, err))
|
||
}
|
||
return options
|
||
}
|
||
|
||
func InitLog() {
|
||
op := newOptions("./conf/stress_1.yaml")
|
||
if err := log.OnInit(op.Setting.Sys["log"]); err != nil {
|
||
panic(fmt.Sprintf("Sys log Init err:%v", err))
|
||
} else {
|
||
log.Infof("stress Sys log Init success !")
|
||
}
|
||
}
|
||
|
||
// ReceivingResults 接收结果并处理
|
||
// 统计的时间都是纳秒，显示的时间 都是毫秒
|
||
// concurrent 并发数
|
||
func ReceivingResults(concurrent uint64, ch <-chan *model.RequestResults, wg *sync.WaitGroup) {
|
||
defer func() {
|
||
wg.Done()
|
||
}()
|
||
|
||
var stopChan = make(chan bool)
|
||
// 时间
|
||
var (
|
||
processingTime uint64 // 处理总时间
|
||
requestTime uint64 // 请求总时间
|
||
maxTime uint64 // 最大时长
|
||
minTime uint64 // 最小时长
|
||
curTime uint64 // 当前消耗时常
|
||
successNum uint64 // 成功处理数，code为0
|
||
failureNum uint64 // 处理失败数，code不为0 0
|
||
chanIDLen int // 并发数
|
||
chanIDs = make(map[uint64]bool)
|
||
receivedBytes int64
|
||
mutex = sync.RWMutex{}
|
||
)
|
||
statTime := uint64(time.Now().UnixNano())
|
||
// 错误码/错误个数
|
||
var errCode = &sync.Map{}
|
||
// 定时输出一次计算结果
|
||
ticker := time.NewTicker(exportStatisticsTime)
|
||
go func() {
|
||
for {
|
||
select {
|
||
case <-ticker.C:
|
||
endTime := uint64(time.Now().UnixNano())
|
||
mutex.Lock()
|
||
go calculateData(concurrent, processingTime, endTime-statTime, maxTime, minTime, successNum, failureNum,
|
||
chanIDLen, errCode, receivedBytes, curTime)
|
||
mutex.Unlock()
|
||
|
||
case <-stopChan:
|
||
// 处理完成
|
||
return
|
||
}
|
||
}
|
||
}()
|
||
header()
|
||
for data := range ch {
|
||
mutex.Lock()
|
||
// fmt.Println("处理一条数据", data.ID, data.Time, data.IsSucceed, data.ErrCode)
|
||
curTime = data.Time
|
||
processingTime = processingTime + data.Time
|
||
if maxTime <= data.Time {
|
||
maxTime = data.Time
|
||
}
|
||
if minTime == 0 {
|
||
minTime = data.Time
|
||
} else if minTime > data.Time {
|
||
minTime = data.Time
|
||
}
|
||
// 是否请求成功
|
||
if data.IsSucceed == true {
|
||
successNum = successNum + 1
|
||
} else {
|
||
failureNum = failureNum + 1
|
||
}
|
||
// 统计错误码
|
||
if value, ok := errCode.Load(data.ErrCode); ok {
|
||
valueInt, _ := value.(int)
|
||
errCode.Store(data.ErrCode, valueInt+1)
|
||
} else {
|
||
errCode.Store(data.ErrCode, 1)
|
||
}
|
||
receivedBytes += data.ReceivedBytes
|
||
if _, ok := chanIDs[data.ChanID]; !ok {
|
||
chanIDs[data.ChanID] = true
|
||
chanIDLen = len(chanIDs)
|
||
}
|
||
requestTimeList = append(requestTimeList, data.Time)
|
||
mutex.Unlock()
|
||
|
||
}
|
||
// 数据全部接受完成，停止定时输出统计数据
|
||
stopChan <- true
|
||
endTime := uint64(time.Now().UnixNano())
|
||
requestTime = endTime - statTime
|
||
calculateData(concurrent, processingTime, requestTime, maxTime, minTime, successNum, failureNum, chanIDLen, errCode,
|
||
receivedBytes, curTime)
|
||
|
||
log.Infof("\n\n")
|
||
log.Infof("************************* 结果 stat ****************************")
|
||
log.Infof("处理协程数量:%d", concurrent)
|
||
|
||
// fmt.Println("处理协程数量:", concurrent, "程序处理总时长:", log("%.3f", float64(processingTime/concurrent)/1e9), "秒")
|
||
log.Infof(fmt.Sprintf("请求总数（并发数*请求数 -c * -n):%d, 总请求时间:%s,秒, successNum:%d,failureNum:%d", successNum+failureNum, fmt.Sprintf("%.3f", float64(requestTime)/1e9), successNum, failureNum))
|
||
printTop(requestTimeList)
|
||
log.Infof("************************* 结果 end ****************************")
|
||
log.Infof("\n\n")
|
||
}
|
||
|
||
// printTop 排序后计算 top 90 95 99
|
||
func printTop(requestTimeList []uint64) {
|
||
if requestTimeList == nil {
|
||
return
|
||
}
|
||
all := tools.MyUint64List{}
|
||
all = requestTimeList
|
||
sort.Sort(all)
|
||
// log.Infof("tp90:", fmt.Sprintf("%.3f", float64(all[int(float64(len(all))*0.90)]/1e6)))
|
||
// log.Infof("tp95:", fmt.Sprintf("%.3f", float64(all[int(float64(len(all))*0.95)]/1e6)))
|
||
// log.Infof("tp99:", fmt.Sprintf("%.3f", float64(all[int(float64(len(all))*0.99)]/1e6)))
|
||
|
||
}
|
||
|
||
// calculateData 计算数据
|
||
func calculateData(concurrent, processingTime, requestTime, maxTime, minTime, successNum, failureNum uint64,
|
||
chanIDLen int, errCode *sync.Map, receivedBytes int64, curTime uint64) {
|
||
if processingTime == 0 {
|
||
processingTime = 1
|
||
}
|
||
var (
|
||
qps float64
|
||
averageTime float64
|
||
maxTimeFloat float64
|
||
minTimeFloat float64
|
||
requestTimeFloat float64
|
||
)
|
||
// 平均 QPS 成功数*总协程数/总耗时 (每秒)
|
||
if processingTime != 0 {
|
||
qps = float64(successNum*1e9*concurrent) / float64(processingTime)
|
||
}
|
||
// 平均时长 总耗时/总请求数/并发数 纳秒=>毫秒
|
||
if successNum != 0 && concurrent != 0 {
|
||
averageTime = float64(processingTime) / float64(successNum*1e6)
|
||
}
|
||
// 纳秒=>毫秒
|
||
maxTimeFloat = float64(maxTime) / 1e6
|
||
minTimeFloat = float64(minTime) / 1e6
|
||
requestTimeFloat = float64(requestTime) / 1e9 // float64(curTime) / 1e6 //
|
||
// 打印的时长都为毫秒
|
||
table(successNum, failureNum, errCode, qps, averageTime, maxTimeFloat, minTimeFloat, requestTimeFloat, chanIDLen,
|
||
receivedBytes)
|
||
}
|
||
|
||
// header 打印表头信息
|
||
func header() {
|
||
log.Infof("\n\n")
|
||
// 打印的时长都为毫秒 总请数
|
||
log.Infof(" ─────┬───────┬───────┬───────┬────────┬────────┬────────┬────────┬────────┬────────┬────────")
|
||
log.Infof(" 耗时│ 并发数│ 成功数│ 失败数│ qps │最长耗时 │最短耗时│平均耗时│下载字节│字节每秒 │ 状态码")
|
||
log.Infof(" ─────┼───────┼───────┼───────┼────────┼────────┼────────┼────────┼────────┼────────┼────────")
|
||
return
|
||
|
||
}
|
||
|
||
// table 打印表格
|
||
func table(successNum, failureNum uint64, errCode *sync.Map,
|
||
qps, averageTime, maxTimeFloat, minTimeFloat, requestTimeFloat float64, chanIDLen int, receivedBytes int64) {
|
||
var (
|
||
speed int64
|
||
)
|
||
if averageTime > 0 {
|
||
speed = int64(float64(receivedBytes) / averageTime * 1e6 / 8 / 1024)
|
||
} else {
|
||
speed = 0
|
||
}
|
||
var (
|
||
receivedBytesStr string
|
||
speedStr string
|
||
)
|
||
// 判断获取下载字节长度是否是未知
|
||
if receivedBytes <= 0 {
|
||
receivedBytesStr = ""
|
||
speedStr = ""
|
||
} else {
|
||
receivedBytesStr = p.Sprintf("%d", receivedBytes)
|
||
speedStr = p.Sprintf("%d", speed)
|
||
}
|
||
// 打印的时长都为毫秒
|
||
result := fmt.Sprintf("%4.0fs│%7d│%7d│%7d│%8.2f│%8.2f│%8.2f│%8.2f│%8s│%8s│%v",
|
||
requestTimeFloat, chanIDLen, successNum, failureNum, qps, maxTimeFloat, minTimeFloat, averageTime,
|
||
receivedBytesStr, speedStr,
|
||
printMap(errCode))
|
||
log.Infof(result)
|
||
return
|
||
}
|
||
|
||
// printMap 输出错误码、次数 节约字符(终端一行字符大小有限)
|
||
func printMap(errCode *sync.Map) (mapStr string) {
|
||
var (
|
||
mapArr []string
|
||
)
|
||
errCode.Range(func(key, value interface{}) bool {
|
||
mapArr = append(mapArr, fmt.Sprintf("%v:%v", key, value))
|
||
return true
|
||
})
|
||
sort.Strings(mapArr)
|
||
mapStr = strings.Join(mapArr, ";")
|
||
return
|
||
}
|