优化
This commit is contained in:
1340691923@qq.com
parent
1d3687dee2
commit
4b2c97d563
@ -115,6 +115,7 @@ func main() {
|
||||
var json = jsoniter.ConfigCompatibleWithStandardLibrary
|
||||
|
||||
err = realTimeDataSarama.Init(model.GlobConfig.Comm.Kafka, model.GlobConfig.Comm.Kafka.ReportTopicName, model.GlobConfig.Comm.Kafka.RealTimeDataGroup, func(msg model.InputMessage, markFn func()) {
|
||||
|
||||
//ETL
|
||||
var kafkaData model.KafkaData
|
||||
err = json.Unmarshal(msg.Value, &kafkaData)
|
||||
|
||||
@ -11,43 +11,28 @@ import (
|
||||
"github.com/1340691923/xwl_bi/platform-basic-libs/sinker"
|
||||
parser "github.com/1340691923/xwl_bi/platform-basic-libs/sinker/parse"
|
||||
"github.com/1340691923/xwl_bi/platform-basic-libs/util"
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
"github.com/tidwall/gjson"
|
||||
"github.com/valyala/fasthttp"
|
||||
"go.uber.org/zap"
|
||||
"math"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
|
||||
|
||||
type ReportController struct {
|
||||
BaseController
|
||||
}
|
||||
|
||||
var parserPool *parser.Pool
|
||||
|
||||
var reportTypeDataPool *sync.Pool
|
||||
|
||||
var Marshaler func(v interface{}) ([]byte, error)
|
||||
|
||||
func init() {
|
||||
func init(){
|
||||
var err error
|
||||
parserPool, err = parser.NewParserPool("fastjson")
|
||||
if err != nil {
|
||||
parserPool,err = parser.NewParserPool("fastjson")
|
||||
if err!=nil{
|
||||
panic(err)
|
||||
}
|
||||
reportTypeDataPool = new(sync.Pool)
|
||||
var json = jsoniter.ConfigCompatibleWithStandardLibrary
|
||||
Marshaler = json.Marshal
|
||||
}
|
||||
|
||||
func GetReportTypeDataPool()*report.ReportTypeData{
|
||||
v := reportTypeDataPool.Get()
|
||||
if reportTypeDataPool.Get() != nil{
|
||||
return new(report.ReportTypeData)
|
||||
}
|
||||
return v.(*report.ReportTypeData)
|
||||
}
|
||||
|
||||
//上报接口
|
||||
@ -59,41 +44,25 @@ func (this ReportController) ReportAction(ctx *fasthttp.RequestCtx) {
|
||||
|
||||
var (
|
||||
typ = ctx.UserValue("typ").(string)
|
||||
appid = ctx.UserValue("appid").(string)
|
||||
appkey = ctx.UserValue("appkey").(string)
|
||||
err error
|
||||
debug = ctx.UserValue("debug").(string)
|
||||
eventName = ctx.UserValue("eventName").(string)
|
||||
body = ctx.Request.Body()
|
||||
)
|
||||
|
||||
reportTypeData := GetReportTypeDataPool()
|
||||
|
||||
reportTypeData.Appid = ctx.UserValue("appid").(string)
|
||||
reportTypeData.Debug = ctx.UserValue("debug").(string)
|
||||
reportTypeData.EventName = ctx.UserValue("eventName").(string)
|
||||
reportTypeData.Body = ctx.Request.Body()
|
||||
|
||||
defer func() {
|
||||
reportTypeData.Appid = ""
|
||||
reportTypeData.TableId = ""
|
||||
reportTypeData.TimeNow = ""
|
||||
reportTypeData.Debug = ""
|
||||
reportTypeData.EventName = ""
|
||||
reportTypeData.Ip = ""
|
||||
reportTypeData.Body = nil
|
||||
reportTypeDataPool.Put(reportTypeData)
|
||||
}()
|
||||
|
||||
if strings.TrimSpace(reportTypeData.EventName) == "" {
|
||||
if strings.TrimSpace(eventName) == ""{
|
||||
this.FastError(ctx, errors.New("事件名 不能为空"))
|
||||
return
|
||||
}
|
||||
|
||||
if strings.TrimSpace(reportTypeData.Appid) == "" {
|
||||
if strings.TrimSpace(appid) == ""{
|
||||
this.FastError(ctx, errors.New("appid 不能为空"))
|
||||
return
|
||||
}
|
||||
|
||||
reportService := report.ReportService{}
|
||||
|
||||
reportTypeData.TableId, err = reportService.GetTableid(reportTypeData.Appid, appkey)
|
||||
tableId, err := reportService.GetTableid(appid, appkey)
|
||||
if err != nil {
|
||||
this.FastError(ctx, err)
|
||||
return
|
||||
@ -106,34 +75,35 @@ func (this ReportController) ReportAction(ctx *fasthttp.RequestCtx) {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
defer duck.Put()
|
||||
|
||||
gjsonArr := gjson.GetManyBytes(reportTypeData.Body, "xwl_distinct_id", "xwl_ip", "xwl_part_date")
|
||||
gjsonArr := gjson.GetManyBytes(body, "xwl_distinct_id", "xwl_ip", "xwl_part_date")
|
||||
|
||||
xwlDistinctId := gjsonArr[0].String()
|
||||
reportTypeData.Ip = gjsonArr[1].String()
|
||||
reportTypeData.TimeNow = gjsonArr[2].String()
|
||||
xwlIp := gjsonArr[1].String()
|
||||
xwlPartDate := gjsonArr[2].String()
|
||||
if xwlDistinctId == "" {
|
||||
this.FastError(ctx, errors.New("xwl_distinct_id 不能为空"))
|
||||
return
|
||||
}
|
||||
|
||||
if reportTypeData.Ip == "" {
|
||||
reportTypeData.Ip = util.CtxClientIP(ctx)
|
||||
if xwlIp == "" {
|
||||
xwlIp = util.CtxClientIP(ctx)
|
||||
}
|
||||
|
||||
if reportTypeData.TimeNow == "" {
|
||||
reportTypeData.TimeNow = time.Now().Format(util.TimeFormat)
|
||||
if xwlPartDate == "" {
|
||||
xwlPartDate = time.Now().Format(util.TimeFormat)
|
||||
}
|
||||
|
||||
duck.NewReportType(reportTypeData)
|
||||
|
||||
duck.NewReportType(appid, tableId, debug, xwlPartDate, eventName, xwlIp, ctx.PostBody())
|
||||
|
||||
if reportService.IsDebugUser(debug, xwlDistinctId, tableId) {
|
||||
kafkaData := duck.GetkafkaData()
|
||||
|
||||
if reportService.IsDebugUser(reportTypeData.Debug, xwlDistinctId, reportTypeData.TableId) {
|
||||
pool := parserPool.Get()
|
||||
defer parserPool.Put(pool)
|
||||
metric, debugErr := pool.Parse(reportTypeData.Body)
|
||||
metric, debugErr := pool.Parse(body)
|
||||
|
||||
if debugErr != nil {
|
||||
logs.Logger.Error("parser.ParseKafkaData ", zap.Error(err))
|
||||
@ -141,7 +111,7 @@ func (this ReportController) ReportAction(ctx *fasthttp.RequestCtx) {
|
||||
return
|
||||
}
|
||||
|
||||
dims, err := sinker.GetDims(model.GlobConfig.Comm.ClickHouse.DbName, kafkaData.GetTableName(), []string{}, db.ClickHouseSqlx, true)
|
||||
dims, err := sinker.GetDims(model.GlobConfig.Comm.ClickHouse.DbName, kafkaData.GetTableName(), []string{}, db.ClickHouseSqlx,true)
|
||||
if err != nil {
|
||||
logs.Logger.Error("sinker.GetDims", zap.Error(err))
|
||||
this.FastError(ctx, errors.New("服务异常"))
|
||||
@ -173,7 +143,7 @@ func (this ReportController) ReportAction(ctx *fasthttp.RequestCtx) {
|
||||
}
|
||||
}
|
||||
|
||||
xwlUpdateTime := gjson.GetBytes(reportTypeData.Body, "xwl_update_time").String()
|
||||
xwlUpdateTime := gjson.GetBytes(body, "xwl_update_time").String()
|
||||
clinetT := util.Str2Time(xwlUpdateTime, util.TimeFormat)
|
||||
serverT := util.Str2Time(kafkaData.ReportTime, util.TimeFormat)
|
||||
if math.Abs(serverT.Sub(clinetT).Minutes()) > 10 {
|
||||
@ -185,7 +155,7 @@ func (this ReportController) ReportAction(ctx *fasthttp.RequestCtx) {
|
||||
m["data_judge"] = "数据检验通过"
|
||||
}
|
||||
|
||||
err = reportService.InflowOfDebugData(m, reportTypeData.EventName)
|
||||
err = reportService.InflowOfDebugData(m, eventName)
|
||||
|
||||
if err != nil {
|
||||
logs.Logger.Error("reportService.InflowOfDebugData", zap.Error(err))
|
||||
@ -198,7 +168,7 @@ func (this ReportController) ReportAction(ctx *fasthttp.RequestCtx) {
|
||||
this.FastError(ctx, my_error.NewError(m["error_reason"].(string), 10006))
|
||||
return
|
||||
}
|
||||
if reportTypeData.Debug == report.DebugNotToDB {
|
||||
if debug == report.DebugNotToDB {
|
||||
this.Output(ctx, map[string]interface{}{
|
||||
"code": 0,
|
||||
"msg": "上报成功（数据不入库）",
|
||||
@ -207,7 +177,7 @@ func (this ReportController) ReportAction(ctx *fasthttp.RequestCtx) {
|
||||
}
|
||||
}
|
||||
|
||||
err = duck.InflowOfKakfa(Marshaler)
|
||||
err = duck.InflowOfKakfa()
|
||||
if err != nil {
|
||||
this.FastError(ctx, err)
|
||||
return
|
||||
|
||||
@ -7,13 +7,14 @@ import (
|
||||
"sync"
|
||||
|
||||
"github.com/Shopify/sarama"
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ReportInterface interface {
|
||||
NewReportType(data *ReportTypeData)
|
||||
NewReportType(appid, tableId, debug, timeNow, eventName, ip string, body []byte)
|
||||
GetkafkaData() model.KafkaData
|
||||
InflowOfKakfa(marshaler func(v interface{}) ([]byte, error)) (err error)
|
||||
InflowOfKakfa() (err error)
|
||||
Put()
|
||||
}
|
||||
|
||||
@ -33,23 +34,13 @@ var eventPool = sync.Pool{
|
||||
},
|
||||
}
|
||||
|
||||
type ReportTypeData struct {
|
||||
Appid string
|
||||
TableId string
|
||||
Debug string
|
||||
TimeNow string
|
||||
EventName string
|
||||
Ip string
|
||||
Body []byte
|
||||
}
|
||||
|
||||
func (this *UserReport) NewReportType(data *ReportTypeData) {
|
||||
this.kafkaData.APPID = data.Appid
|
||||
this.kafkaData.TableId = data.TableId
|
||||
this.kafkaData.Debug = data.Debug
|
||||
this.kafkaData.ReqData = data.Body
|
||||
this.kafkaData.Ip = data.Ip
|
||||
this.kafkaData.ReportTime = data.TimeNow
|
||||
func (this *UserReport) NewReportType(appid, tableId, debug, timeNow, eventName, ip string, body []byte) {
|
||||
this.kafkaData.APPID = appid
|
||||
this.kafkaData.TableId = tableId
|
||||
this.kafkaData.Debug = debug
|
||||
this.kafkaData.ReqData = body
|
||||
this.kafkaData.Ip = ip
|
||||
this.kafkaData.ReportTime = timeNow
|
||||
this.kafkaData.ReportType = model.UserReportType
|
||||
this.kafkaData.EventName = "用户属性"
|
||||
}
|
||||
@ -58,11 +49,12 @@ func (this *UserReport) GetkafkaData() model.KafkaData {
|
||||
return this.kafkaData
|
||||
}
|
||||
|
||||
func (this *UserReport) InflowOfKakfa(marshaler func(v interface{}) ([]byte, error)) (err error) {
|
||||
func (this *UserReport) InflowOfKakfa() (err error) {
|
||||
|
||||
var json = jsoniter.ConfigCompatibleWithStandardLibrary
|
||||
msg := &sarama.ProducerMessage{}
|
||||
msg.Topic = model.GlobConfig.Comm.Kafka.ReportTopicName
|
||||
sendData, _ := marshaler(this.kafkaData)
|
||||
sendData, _ := json.Marshal(this.kafkaData)
|
||||
msg.Value = sarama.ByteEncoder(sendData)
|
||||
msg.Timestamp = time.Now()
|
||||
|
||||
@ -77,23 +69,22 @@ type EventReport struct {
|
||||
kafkaData model.KafkaData
|
||||
}
|
||||
|
||||
func (this *EventReport) NewReportType(data *ReportTypeData) {
|
||||
this.kafkaData.APPID = data.Appid
|
||||
this.kafkaData.TableId = data.TableId
|
||||
this.kafkaData.Debug = data.Debug
|
||||
this.kafkaData.ReqData = data.Body
|
||||
this.kafkaData.ReportTime = data.TimeNow
|
||||
func (this *EventReport) NewReportType(appid, tableId, debug, timeNow, eventName, ip string, body []byte) {
|
||||
this.kafkaData.APPID = appid
|
||||
this.kafkaData.TableId = tableId
|
||||
this.kafkaData.Debug = debug
|
||||
this.kafkaData.ReqData = body
|
||||
this.kafkaData.ReportTime = timeNow
|
||||
this.kafkaData.ReportType = model.EventReportType
|
||||
this.kafkaData.EventName = data.EventName
|
||||
this.kafkaData.Ip = data.Ip
|
||||
this.kafkaData.EventName = eventName
|
||||
this.kafkaData.Ip = ip
|
||||
}
|
||||
|
||||
func (this *EventReport) InflowOfKakfa(marshaler func(v interface{}) ([]byte, error)) (err error) {
|
||||
|
||||
func (this *EventReport) InflowOfKakfa() (err error) {
|
||||
var json = jsoniter.ConfigCompatibleWithStandardLibrary
|
||||
msg := &sarama.ProducerMessage{}
|
||||
msg.Topic = model.GlobConfig.Comm.Kafka.ReportTopicName
|
||||
sendData, _ := marshaler(this.kafkaData)
|
||||
|
||||
sendData, _ := json.Marshal(this.kafkaData)
|
||||
msg.Value = sarama.ByteEncoder(sendData)
|
||||
msg.Timestamp = time.Now()
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user