diff --git a/dsDataex/MyService/Auth/AuthService/AuthService.go b/dsDataex/MyService/Auth/AuthService/AuthService.go index 187f1f36..ac4a90e3 100644 --- a/dsDataex/MyService/Auth/AuthService/AuthService.go +++ b/dsDataex/MyService/Auth/AuthService/AuthService.go @@ -70,6 +70,28 @@ func CheckToken(systemID string,authToken string) (bool, string,string){ return true,"接入系统票据验证成功",data["id"].(string) }else { + return false,"接入系统票据验证失败","" + } + } +} + +func CheckAccessToken(systemID string,accessTime string,accessToken string) (bool, string,string){ + + success,result,data,_:= AuthDAO.GetbyCode(systemID) + + if success==false{ + + return false,result,"" + }else { + myToken:= data["auth_token"].(string) + + var testToken=MD5Util.MD5V1(systemID + accessTime + myToken) + + if testToken==accessToken{ + + return true,"接入系统票据验证成功",data["id"].(string) + }else { + return false,"接入系统票据验证失败","" } } diff --git a/dsDataex/MyService/DataEX/DataexOpenapi/DataexOpenapi.go b/dsDataex/MyService/DataEX/DataexOpenapi/DataexOpenapi.go index 76e458bf..709914f9 100644 --- a/dsDataex/MyService/DataEX/DataexOpenapi/DataexOpenapi.go +++ b/dsDataex/MyService/DataEX/DataexOpenapi/DataexOpenapi.go @@ -6,6 +6,7 @@ import ( "dsDataex/MyService/MySwagger" "github.com/gin-gonic/gin" "net/http" + "strings" ) // @Summary 修改数据 @@ -86,19 +87,60 @@ func DataexCollect(c *gin.Context) { var input MySwagger.DataCollect if err := c.ShouldBindJSON(&input); err != nil { - c.JSON(http.StatusBadRequest, MySwagger.Result{Success: false,Message: "接入系统数据JSON格式错误"}) + c.JSON(http.StatusOK, MySwagger.Result{Success: false,Message: "接入系统数据JSON格式错误"}) return } if input.AccessToken == "" { - c.JSON(http.StatusUnauthorized, MySwagger.Result{Success: false,Message: "接入系统票据验证失败"}) + c.JSON(http.StatusOK, MySwagger.Result{Success: false,Message: "接入系统票据验证失败"}) + return + } + + if len(input.EventDatas)==0 { + c.JSON(http.StatusOK, MySwagger.Result{Success: false,Message: "接入系统JSON数据EventDatas不能为空"}) + return + } + + var temp=strings.Split(input.AccessToken,"##") + + if len(temp)!=3{ + c.JSON(http.StatusOK, MySwagger.Result{Success: false,Message: "接入系统票据格式不正确"}) + return + } + + flag,_,systemID:=AuthService.CheckAccessToken(temp[0],temp[1],temp[2]) + + if flag==false { + c.JSON(http.StatusOK, MySwagger.Result{Success: false,Message: "接入系统票据不正确"}) + return + } + + flag2,_,dataSource := DataexService.CheckDatasourceSet(systemID,input.DataSource) + + if flag2==false { + c.JSON(http.StatusOK, MySwagger.Result{Success: false,Message: "接入系统访问的数据源类型不正确"}) return } - c.JSON(http.StatusOK, MySwagger.Result{ - Success: true, - Message: "DataexCollect服务调用成功", - }) + var ip = c.ClientIP() + + success,message:=DataexService.DataexCollect(temp[0],input.UserDatas,input.EventDatas,dataSource,ip) + + if success{ + c.JSON(http.StatusOK, MySwagger.Result{ + Success: true, + Message: "DataexCollect服务调用成功", + }) + + return + }else { + c.JSON(http.StatusOK, MySwagger.Result{ + Success: false, + Message: message, + }) + + return + } return } diff --git a/dsDataex/MyService/DataEX/DataexService/DataexService.go b/dsDataex/MyService/DataEX/DataexService/DataexService.go index b0cdbbe6..7116b9d3 100644 --- a/dsDataex/MyService/DataEX/DataexService/DataexService.go +++ b/dsDataex/MyService/DataEX/DataexService/DataexService.go @@ -7,6 +7,7 @@ import ( "dsDataex/MyService/MySwagger" "dsDataex/Utils/CacheUtil" "dsDataex/Utils/ES7Util" + "dsDataex/Utils/KafkaUtil" "encoding/json" "strconv" "time" @@ -188,6 +189,114 @@ func DataexSet(systemID string, datas []MySwagger.Data,datasource *models.TDatae return false,"数据源类型配置不正确",nil,nil } +func DataexCollect(systemID string,users []MySwagger.User,events []MySwagger.Event,datasource *models.TDataexDatasource,ip string) (bool,string) { + + var now = time.Now().Format("2006/01/02 15:04:05") + + var userInfo = make( map[string]MySwagger.User) + var accessInfo = make( map[string]MySwagger.User) + + if len(users)>0{ + for no:=0;no< len(users);no++{ + if users[no].UserID==""{ + _,flag:= accessInfo[users[no].AccessID] + + if flag==false{ + accessInfo[users[no].AccessID]=users[no] + } + }else { + _,flag:= userInfo[users[no].UserID] + + if flag==false{ + userInfo[users[no].UserID]=users[no] + } + } + } + } + + var datas []DataEX.KafkaData + + for no:=0;no< len(events);no++{ + + var data DataEX.KafkaData + + if events[no].EventUserID==""{ + if len(users)>0{ + data.UserID=users[0].UserID + data.AccessID=users[0].AccessID + data.Identity=users[0].Identity + data.AccessWay=users[0].AccessWay + data.AccessIP=users[0].AccessIP + + var jsonData map[string]interface{} + json.Unmarshal([]byte(users[0].Properties), &jsonData) + + data.UserProperty =jsonData + } + }else { + _,flag:= userInfo[events[no].EventUserID] + + if flag==true{ + data.UserID=userInfo[events[no].EventUserID].UserID + data.AccessID=userInfo[events[no].EventUserID].AccessID + data.Identity=userInfo[events[no].EventUserID].Identity + data.AccessWay=userInfo[events[no].EventUserID].AccessWay + data.AccessIP=userInfo[events[no].EventUserID].AccessIP + + var jsonData map[string]interface{} + json.Unmarshal([]byte(userInfo[events[no].EventUserID].Properties), &jsonData) + + data.UserProperty =jsonData + } + + _,flag2:= accessInfo[events[no].EventUserID] + + if flag2==true{ + data.UserID=accessInfo[events[no].EventUserID].UserID + data.AccessID=accessInfo[events[no].EventUserID].AccessID + data.Identity=accessInfo[events[no].EventUserID].Identity + data.AccessWay=accessInfo[events[no].EventUserID].AccessWay + data.AccessIP=accessInfo[events[no].EventUserID].AccessIP + + var jsonData map[string]interface{} + json.Unmarshal([]byte(accessInfo[events[no].EventUserID].Properties), &jsonData) + + data.UserProperty =jsonData + } + } + + data.EventType=events[no].EventType + data.EventName=events[no].EventName + data.EventTime=events[no].EventTime + data.EventURI=events[no].EventURI + data.EventSeqNO=events[no].EventSeqNO + + var jsonData map[string]interface{} + json.Unmarshal([]byte(events[no].Properties), &jsonData) + + data.EventProperty=jsonData + + data.SystemId=systemID + data.DatasourceId=datasource.DatasourceCode + data.CollectTime=now + + if data.AccessIP==""{ + data.AccessIP=ip + } + + datas=append(datas,data) + } + + //flag,result:=KafkaUtil.Provide(datasource.DatasourceCode,datas) + flag,result:=KafkaUtil.ProvideLow(datasource.DatasourceCode,datas) + + if flag==true{ + return true,"数据汇集成功" + }else { + return false,result + } +} + /** * @Author zhangjun * @Description Set索引文档，批量发送 @@ -204,8 +313,6 @@ func DataexSetBatch(systemID string, datas []MySwagger.Data,datasource *models.T //add by zhangjun 2020-07-15 //处理组织机构树同步 if datasource.DatasourceCode=="org_school" && systemID=="BASE_GO"{ - //change by zhangjun 2020-07-16 - //OrgtreeProc(datas) OrgtreeProcBatch(datas) } if datasource.DatasourceCode=="org_school_lua" && systemID=="BASE_LUA"{ diff --git a/dsDataex/MyService/DataEX/KafkaData.go b/dsDataex/MyService/DataEX/KafkaData.go new file mode 100644 index 00000000..a2978a4f --- /dev/null +++ b/dsDataex/MyService/DataEX/KafkaData.go @@ -0,0 +1,23 @@ +package DataEX + +type KafkaData struct { + + SystemId string `json:"system_id"` + DatasourceId string `json:"datasource_id"` + + UserID string `json:"user_id"` + Identity string `json:"identity"` + AccessID string `json:"access_id"` + AccessIP string `json:"access_ip"` + AccessWay string `json:"access_way"` + + EventType string `json:"event_type"` + EventName string `json:"event_name"` + EventTime string `json:"event_time"` + EventURI string `json:"event_uri"` + EventSeqNO string `json:"event_seqno"` + CollectTime string `json:"collect_time"` + + UserProperty map[string]interface{} `json:"user_property"` + EventProperty map[string]interface{} `json:"event_property"` +} diff --git a/dsDataex/MyService/MySwagger/DataCollect.go b/dsDataex/MyService/MySwagger/DataCollect.go index 0ff446ea..642ae9c2 100644 --- a/dsDataex/MyService/MySwagger/DataCollect.go +++ b/dsDataex/MyService/MySwagger/DataCollect.go @@ -4,9 +4,7 @@ type DataCollect struct { AccessToken string `json:"access_token" example:"system_01##20200102030405##a6ce11eab94df48a6ce11eab" ` DataSource string `json:"data_source" example:"log_login" ` - AccessIP string `json:"access_ip" example:"10.10.8.88"` - AccessWay string `json:"access_way" example:"app/web/weixin/dingding"` - UserData User `json:"user_data" ` + UserDatas []User `json:"user_datas" ` EventDatas []Event `json:"event_datas" ` } \ No newline at end of file diff --git a/dsDataex/MyService/MySwagger/Event.go b/dsDataex/MyService/MySwagger/Event.go index 4ab0efde..8454db2a 100644 --- a/dsDataex/MyService/MySwagger/Event.go +++ b/dsDataex/MyService/MySwagger/Event.go @@ -5,9 +5,9 @@ type Event struct { EventType string `json:"event_type" example:"1001"` EventName string `json:"event_name" example:"在线提交作业"` EventTime string `json:"event_time" example:"2020-08-08 18:19:20"` - EventURI string `json:"event_uri" example:"https://doc.thinkingdata.cn/user_guide"` + EventURI string `json:"event_uri" example:"https://edusoa.com/cloud/study/do_homework"` EventSeqNO string `json:"event_seqno" example:"事件顺序号：1/2/3"` - EventUserID string `json:"event_userid" example:"用户ID(可选)"` + EventUserID string `json:"event_userid" example:"用户ID"` Properties string `json:"properties" example:"{res_id:12300,res_name:初一语文一单元作业,teacher_id:123001,time_spend:100s}"` } diff --git a/dsDataex/MyService/MySwagger/User.go b/dsDataex/MyService/MySwagger/User.go index b1b892f5..5bd57dfc 100644 --- a/dsDataex/MyService/MySwagger/User.go +++ b/dsDataex/MyService/MySwagger/User.go @@ -4,7 +4,9 @@ type User struct { UserID string `json:"user_id" example:"用户ID"` Identity string `json:"identity" example:"用户身份（1：教师，2：学生，3：家长，4：管理员，5：访客）"` - AccessID string `json:"access_id" example:"访客ID（可选）"` + AccessID string `json:"access_id" example:"访客ID（用户未登录）"` + AccessIP string `json:"access_ip" example:"10.10.8.88"` + AccessWay string `json:"access_way" example:"app/web/weixin/dingding"` Properties string `json:"properties" example:"{group_id:12300,group_name:课外学习一组,user_level:2,login_time:2020/08/08 12:13:14}"` } diff --git a/dsDataex/Utils/ES7Util/ES7Util.go b/dsDataex/Utils/ES7Util/ES7Util.go index c93b8b45..fecb0953 100644 --- a/dsDataex/Utils/ES7Util/ES7Util.go +++ b/dsDataex/Utils/ES7Util/ES7Util.go @@ -35,7 +35,7 @@ func init() { //Ping，获取ES server启动信息 info, _, _ := ES7Client.Ping(ConfigUtil.ESAddress[0]).Do(CTX) - ServerVersion = info.Version.Number + ServerVersion =ES7Client.String() +"【"+ info.Version.Number+"】" } /** diff --git a/dsDataex/Utils/KafkaUtil/KafkaUtil.go b/dsDataex/Utils/KafkaUtil/KafkaUtil.go index 3dc20a0e..7129c59b 100644 --- a/dsDataex/Utils/KafkaUtil/KafkaUtil.go +++ b/dsDataex/Utils/KafkaUtil/KafkaUtil.go @@ -3,46 +3,125 @@ package KafkaUtil import ( "bytes" "context" + "dsDataex/MyService/DataEX" "dsDataex/Utils/ConfigUtil" + "encoding/json" "fmt" "github.com/segmentio/kafka-go" + "math/rand" "strconv" "time" ) -var KafkaClient *kafka.Conn var KafkaBroker string var CTX context.Context +var writerPool map[string]*kafka.Writer +var kafkaPool map[string]map[int]*kafka.Conn +var kafkaParts map[string]int + func init() { CTX=context.Background() - KafkaClient, _ = kafka.DialLeader(CTX , "tcp", ConfigUtil.KafkaBrokers[0], "__consumer_offsets" , 0) + KafkaClient, _ := kafka.DialLeader(CTX , "tcp", ConfigUtil.KafkaBrokers[0], "__consumer_offsets" , 0) brokers,_:= KafkaClient.Brokers() - KafkaBroker=brokers[0].Host+":"+strconv.Itoa(brokers[0].Port)+"【"+strconv.Itoa(brokers[0].ID)+"】" + writerPool=make(map[string]*kafka.Writer) + + kafkaPool = make(map[string]map[int]*kafka.Conn) + kafkaParts= make(map[string]int) } -func ProvideLow(topic string) { +/** + * @Author zhangjun + * @Description + * @Date 2020-07-27 05:26 + * @Param + * @return + **/ +func ProvideLow(topic string,datas []DataEX.KafkaData)(bool,string){ - KafkaClient, _ = kafka.DialLeader(context.Background(), "tcp", ConfigUtil.KafkaBrokers[0], topic , 0) + var client *kafka.Conn + clients,flag:=kafkaPool[topic] - KafkaClient.SetWriteDeadline(time.Now().Add(5 *time.Second)) + if flag==false{ - KafkaClient.WriteMessages( - kafka.Message{Value: []byte("one!")}, - kafka.Message{Value: []byte("two!")}, - kafka.Message{Value: []byte("three!")}, - ) + client,_= kafka.DialLeader(CTX, "tcp", ConfigUtil.KafkaBrokers[0], topic , 0) - //KafkaClient.Close() + clients=make(map[int]*kafka.Conn) + clients[0]=client + + parts,_:=client.ReadPartitions() + offset,_:=client.ReadLastOffset() + + if len(parts)== 1 && offset==0 {//初始化 Topic + + DeleteTopic(topic) + time.Sleep(100 * time.Millisecond) + CreateTopic(topic) + + client, _ = kafka.DialLeader(CTX, "tcp", ConfigUtil.KafkaBrokers[0], topic , 0) + + clients=make(map[int]*kafka.Conn) + clients[0]=client + + parts,_=client.ReadPartitions() + } + + if len(parts)>1{//TODO:预先加载 Kafka连接池，可能影响性能，暂不实现！！！ + + } + + kafkaPool[topic]=clients + kafkaParts[topic]= len(parts) + + }else { + + max:=kafkaParts[topic] + num:=rand.Intn(max) + c,f:=kafkaPool[topic][num] + + if f==true{ + client=c + }else { + client, _ = kafka.DialLeader(CTX, "tcp", ConfigUtil.KafkaBrokers[0], topic , num) + kafkaPool[topic][num]=client + } + } + + //client, _ := kafka.DialLeader(context.Background(), "tcp", ConfigUtil.KafkaBrokers[0], topic , 0) + + var messages []kafka.Message + + for no:=0;no< len(datas);no++{ + + var data,_=json.Marshal(datas[no]) + + var msg=kafka.Message{Value: data } + + messages=append(messages,msg) + } + + client.SetWriteDeadline(time.Now().Add(5 *time.Second)) + + _,err:= client.WriteMessages(messages...) + + //client.Close() + + if err ==nil{ + + return true,"" + }else { + + return false,"Kafka数据存储错误" + } } func ConsumeLow(topic string) { - KafkaClient, _ = kafka.DialLeader(context.Background(), "tcp", ConfigUtil.KafkaBrokers[0], topic , 0) + KafkaClient, _ := kafka.DialLeader(context.Background(), "tcp", ConfigUtil.KafkaBrokers[0], topic , 0) KafkaClient.SetReadDeadline(time.Now().Add(5 *time.Second)) @@ -89,33 +168,60 @@ func Consume(topic string,group string) { r.Close() } -func Provide(topic string){ +/** + * @Author zhangjun + * @Description + * @Date 2020-07-27 11:13 + * @Param + * @return + * TODO:高阶接口调用速度太慢，1s，不采用！！！ + **/ +func Provide(topic string,datas []DataEX.KafkaData)(bool,string){ - w := kafka.NewWriter(kafka.WriterConfig{ - Brokers: ConfigUtil.KafkaBrokers, - Topic: topic, - Balancer: &kafka.LeastBytes{},//Partition 自动分配器 - }) + var begin =time.Now() + w,f:= writerPool[topic] + + if f==false{ + + w = kafka.NewWriter(kafka.WriterConfig{ + Brokers: ConfigUtil.KafkaBrokers, + Topic: topic, + Balancer: &kafka.Hash{},//.RoundRobin{},//.LeastBytes{}, + }) + + writerPool[topic]=w + } - w.WriteMessages(context.Background(), - kafka.Message{ - //Key: []byte("Key-A"), - Value: []byte("Hello World! One"), - }, - kafka.Message{ - Value: []byte("Hello World! Two"), - }, - kafka.Message{ - Value: []byte("Hello World! Three"), - }, - ) - - w.Close() + var messages []kafka.Message + + for no:=0;no< len(datas);no++{ + + var data,_=json.Marshal(datas[no]) + + var msg=kafka.Message{Value: data } + + messages=append(messages,msg) + } + + fmt.Println("Time 9:",time.Now(),",spend:",time.Since(begin)) + + err:= w.WriteMessages(CTX ,messages...) + + fmt.Println("Time 10:",time.Now(),",spend:",time.Since(begin)) + //w.Close() + + if err ==nil{ + + return true,"" + }else { + + return false,"Kafka数据存储错误" + } } func CreateTopic(topic string){ - KafkaClient, _ = kafka.DialLeader(CTX , "tcp", ConfigUtil.KafkaBrokers[0], "__consumer_offsets" , 0) + KafkaClient, _ := kafka.DialLeader(CTX , "tcp", ConfigUtil.KafkaBrokers[0], "__consumer_offsets" , 0) err:= KafkaClient.CreateTopics(kafka.TopicConfig{ NumPartitions: 8, @@ -129,7 +235,7 @@ func CreateTopic(topic string){ } func DeleteTopic(topic string){ - KafkaClient, _ = kafka.DialLeader( CTX, "tcp", ConfigUtil.KafkaBrokers[0], topic , 0) + KafkaClient, _ := kafka.DialLeader( CTX, "tcp", ConfigUtil.KafkaBrokers[0], topic , 0) err:= KafkaClient.DeleteTopics(topic) diff --git a/dsDataex/docs/docs.go b/dsDataex/docs/docs.go index f70be7c7..cc4c95b4 100644 --- a/dsDataex/docs/docs.go +++ b/dsDataex/docs/docs.go @@ -1450,18 +1450,10 @@ var doc = `{ "MySwagger.DataCollect": { "type": "object", "properties": { - "access_ip": { - "type": "string", - "example": "10.10.8.88" - }, "access_token": { "type": "string", "example": "system_01##20200102030405##a6ce11eab94df48a6ce11eab" }, - "access_way": { - "type": "string", - "example": "app/web/weixin/dingding" - }, "data_source": { "type": "string", "example": "log_login" @@ -1472,8 +1464,11 @@ var doc = `{ "$ref": "#/definitions/MySwagger.Event" } }, - "user_data": { - "$ref": "#/definitions/MySwagger.User" + "user_datas": { + "type": "array", + "items": { + "$ref": "#/definitions/MySwagger.User" + } } } }, @@ -1762,11 +1757,11 @@ var doc = `{ }, "event_uri": { "type": "string", - "example": "https://doc.thinkingdata.cn/user_guide" + "example": "https://edusoa.com/cloud/study/do_homework" }, "event_userid": { "type": "string", - "example": "用户ID(可选)" + "example": "用户ID" }, "properties": { "type": "string", @@ -2136,7 +2131,15 @@ var doc = `{ "properties": { "access_id": { "type": "string", - "example": "访客ID（可选）" + "example": "访客ID（用户未登录）" + }, + "access_ip": { + "type": "string", + "example": "10.10.8.88" + }, + "access_way": { + "type": "string", + "example": "app/web/weixin/dingding" }, "identity": { "type": "string", diff --git a/dsDataex/docs/swagger.json b/dsDataex/docs/swagger.json index 6438c228..194cfd51 100644 --- a/dsDataex/docs/swagger.json +++ b/dsDataex/docs/swagger.json @@ -1435,18 +1435,10 @@ "MySwagger.DataCollect": { "type": "object", "properties": { - "access_ip": { - "type": "string", - "example": "10.10.8.88" - }, "access_token": { "type": "string", "example": "system_01##20200102030405##a6ce11eab94df48a6ce11eab" }, - "access_way": { - "type": "string", - "example": "app/web/weixin/dingding" - }, "data_source": { "type": "string", "example": "log_login" @@ -1457,8 +1449,11 @@ "$ref": "#/definitions/MySwagger.Event" } }, - "user_data": { - "$ref": "#/definitions/MySwagger.User" + "user_datas": { + "type": "array", + "items": { + "$ref": "#/definitions/MySwagger.User" + } } } }, @@ -1747,11 +1742,11 @@ }, "event_uri": { "type": "string", - "example": "https://doc.thinkingdata.cn/user_guide" + "example": "https://edusoa.com/cloud/study/do_homework" }, "event_userid": { "type": "string", - "example": "用户ID(可选)" + "example": "用户ID" }, "properties": { "type": "string", @@ -2121,7 +2116,15 @@ "properties": { "access_id": { "type": "string", - "example": "访客ID（可选）" + "example": "访客ID（用户未登录）" + }, + "access_ip": { + "type": "string", + "example": "10.10.8.88" + }, + "access_way": { + "type": "string", + "example": "app/web/weixin/dingding" }, "identity": { "type": "string", diff --git a/dsDataex/docs/swagger.yaml b/dsDataex/docs/swagger.yaml index 8b8871d4..508a11ba 100644 --- a/dsDataex/docs/swagger.yaml +++ b/dsDataex/docs/swagger.yaml @@ -29,15 +29,9 @@ definitions: type: object MySwagger.DataCollect: properties: - access_ip: - example: 10.10.8.88 - type: string access_token: example: system_01##20200102030405##a6ce11eab94df48a6ce11eab type: string - access_way: - example: app/web/weixin/dingding - type: string data_source: example: log_login type: string @@ -45,8 +39,10 @@ definitions: items: $ref: '#/definitions/MySwagger.Event' type: array - user_data: - $ref: '#/definitions/MySwagger.User' + user_datas: + items: + $ref: '#/definitions/MySwagger.User' + type: array type: object MySwagger.DataGet: properties: @@ -246,10 +242,10 @@ definitions: example: "1001" type: string event_uri: - example: https://doc.thinkingdata.cn/user_guide + example: https://edusoa.com/cloud/study/do_homework type: string event_userid: - example: 用户ID(可选) + example: 用户ID type: string properties: example: '{res_id:12300,res_name:初一语文一单元作业,teacher_id:123001,time_spend:100s}' @@ -500,7 +496,13 @@ definitions: MySwagger.User: properties: access_id: - example: 访客ID（可选） + example: 访客ID（用户未登录） + type: string + access_ip: + example: 10.10.8.88 + type: string + access_way: + example: app/web/weixin/dingding type: string identity: example: 用户身份（1：教师，2：学生，3：家长，4：管理员，5：访客） diff --git a/dsDataex/main.go b/dsDataex/main.go index 155a9825..b082d618 100644 --- a/dsDataex/main.go +++ b/dsDataex/main.go @@ -108,14 +108,14 @@ func GinServerInit() { // @securityDefinitions.basic BasicAuth func main() { - fmt.Println("Dsideal DataEX GO GO GO !!!") + fmt.Println("DsDataEX GO ! GO !! GO !!!") fmt.Println("ES Server :" + ES7Util.ServerVersion) //fmt.Println("ES Server :" + ES7SqlUtil.ServerVersion) - fmt.Println("Kafka Server :" + KafkaUtil.KafkaBroker ) - //KafkaUtil.CreateTopic("log_12345") - //KafkaUtil.DeleteTopic("log_12345") + + //var token=MD5Util.MD5V1("TEST_007" + "20200707" + "DSDataex_Token_7ee1f0f76243449f8d75f40fdcc2b93d") + //fmt.Println("AccessToken : "+token) CacheUtil.OrgtreeCacheInit()