bigdata
/
dsMin
6
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

kafka provide

Browse Source
master
zhangjun 5 years ago
parent 4bc564637d
commit dd454f0fd2
13 changed files with 399 additions and 91 deletions
Show Stats Download Patch File Download Diff File

22
dsDataex/MyService/Auth/AuthService/AuthService.go
Unescape View File

@ -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,"接入系统票据验证失败",""
}
}

54
dsDataex/MyService/DataEX/DataexOpenapi/DataexOpenapi.go
Unescape View File

@ -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
}

111
dsDataex/MyService/DataEX/DataexService/DataexService.go
Unescape View File

@ -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"{

23
dsDataex/MyService/DataEX/KafkaData.go
Unescape View File

@ -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"`
}

4
dsDataex/MyService/MySwagger/DataCollect.go
Unescape View File

@ -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" `
}

4
dsDataex/MyService/MySwagger/Event.go
Unescape View File

@ -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}"`
}

4
dsDataex/MyService/MySwagger/User.go
Unescape View File

@ -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}"`
}

2
dsDataex/Utils/ES7Util/ES7Util.go
Unescape View File

@ -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+"】"
}
/**

176
dsDataex/Utils/KafkaUtil/KafkaUtil.go
Unescape View File

@ -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)

29
dsDataex/docs/docs.go
Unescape View File

@ -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",

29
dsDataex/docs/swagger.json
Unescape View File

@ -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",

24
dsDataex/docs/swagger.yaml
Unescape View File

@ -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：访客）

8
dsDataex/main.go
Unescape View File

@ -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()

Write Preview
Loading…
Cancel
Save