up

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/BurntSushi/toml v0.3.1
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/ichunt2019/go-msgserver v1.0.5 // indirect
+	github.com/ichunt2019/go-rabbitmq v1.0.1 // indirect
 	github.com/ichunt2019/logger v1.0.5
 	github.com/jmoiron/sqlx v1.2.0
 	go.mongodb.org/mongo-driver v1.3.0

go.sum

@@ -36,6 +36,8 @@ github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEW
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/ichunt2019/go-msgserver v1.0.5 h1:yanQ95Ld0etJzVhyZql0jhXXE18qGFX48qFgTCim5hg=
 github.com/ichunt2019/go-msgserver v1.0.5/go.mod h1:fWAvbry0W9nhmkqgT2agwRqYCWlguUJXgy2rgoWOUmA=
+github.com/ichunt2019/go-rabbitmq v1.0.1 h1:qHhpGm9v7jnhSBo3f3viX+BSky9yugp9lCSV03eYsF4=
+github.com/ichunt2019/go-rabbitmq v1.0.1/go.mod h1:TQsZ1XWULyvm4UwpYHwNPtOXYbuVvLLI0GM7g/BRy68=
 github.com/ichunt2019/logger v1.0.5 h1:85C6kJCH9xlbLt1VmwHp/8iScm+bIlenK6nanWwwq/o=
 github.com/ichunt2019/logger v1.0.5/go.mod h1:5IWMrrqJIWwOIGav9ACWOI+KOuYeteUvOei4zubclwg=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
@@ -72,6 +74,8 @@ github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271 h1:WhxRHzgeVGETMlmVfqhRn8RIeeNoPr2Czh33I4Zdccw=
 github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
+github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71 h1:2MR0pKUzlP3SGgj5NYJe/zRYDwOu9ku6YHy+Iw7l5DM=
+github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=

service/behavior/apiMsgService/apiMsgService.go

@@ -2,6 +2,7 @@ package apiMsgService
 
 import (
 	"encoding/json"
+	"time"
 )
 
 // 队列参数
@@ -25,49 +26,8 @@ var MsgParams *DataParams
 //写日志到管道
 func  SaveMsgToLogChan(msg string) (err error) {
 	err = json.Unmarshal([]byte(msg),&MsgParams)
+	MsgParams.CreateTime = time.Now().Unix()
+	MsgParams.CreateTimeStr = time.Now().Format("2006-01-02 15:04:05")
 	G_logSink.Append(MsgParams)
 	return err
 }
-//
-//func SaveElkLog(LogReportDir string){
-//	var(
-//		logBatch *util.LogBatch
-//	)
-//	fileLog,_ := util.NewFileLogger(LogReportDir)
-//	fileLog.Init()
-//
-//	for{
-//		select{
-//			case util.MsgParams =  <- util.MsgChan:
-//				if logBatch == nil{
-//					logBatch = &util.LogBatch{}
-//				}
-//				// 把新日志追加到批次中
-//				logBatch.Logs = append(logBatch.Logs, util.MsgParams)
-//				// 如果批次满了, 就立即写入
-//				if len(logBatch.Logs) > 500{
-//					//写入日志
-//					fmt.Println("长度到了开始写入日志")
-//					defer fileLog.Close()
-//					fileLog.WriteLog(logBatch)
-//					// 清空logBatch
-//					logBatch = nil
-//				}
-//			case <- time.NewTimer(10*time.Second).C:
-//
-//				if logBatch == nil{
-//					logBatch = &util.LogBatch{}
-//				}
-//				fmt.Println("超时到期了")
-//				fmt.Println(len(logBatch.Logs))
-//				//写入日志
-//				fmt.Println("超时了写入日志")
-//				defer fileLog.Close()
-//				fileLog.WriteLog(logBatch)
-//				// 清空logBatch
-//				logBatch = nil
-//		}
-//
-//		time.Sleep(time.Microsecond*10)
-//	}
-//}
\ No newline at end of file

service/behavior/main.go

@@ -2,13 +2,10 @@ package main
 
 import (
 	"flag"
-	"fmt"
-	"github.com/ichunt2019/go-msgserver/utils/rabbitmq"
+	"github.com/ichunt2019/go-rabbitmq/utils/rabbitmq"
 	"github.com/ichunt2019/logger"
 	"go-api-behavior/service/behavior/apiMsgService"
 	"go-api-behavior/util"
-	"sync"
-	"time"
 )
 
 type RecvPro struct {
@@ -57,8 +54,7 @@ func main() {
 
 	//初始化配置文件
 	util.Init(ConfigDir)
-	
-	//util.MsgChan = make(chan *util.DataParams,10)
+
 
 	//
 	logConfig := make(map[string]string)
@@ -71,38 +67,22 @@ func main() {
 
 	t := &RecvPro{}
 
-	queueExchange := &rabbitmq.QueueExchange{
+
+
+	_ = apiMsgService.InitLogSink()
+	
+
+
+
+	rabbitmq.Recv(rabbitmq.QueueExchange{
 		util.Configs.Rabbitmq_ichunt.QueueName,
 		util.Configs.Rabbitmq_ichunt.RoutingKey,
 		util.Configs.Rabbitmq_ichunt.Exchange,
 		util.Configs.Rabbitmq_ichunt.Type,
 		util.Configs.Rabbitmq_ichunt.Dns,
-	}
+	},t,3)
+
+
+
 
-	_ = apiMsgService.InitLogSink()
-	
-	//go func(){
-	//	apiMsgService.SaveElkLog(LogReportDir)
-	//}()
-
-	for{
-		var wg sync.WaitGroup
-		//fmt.Println("开始执行任务....")
-		for i := 0;i<3;i++{
-			wg.Add(1)
-			go func(wg *sync.WaitGroup){
-				mq := rabbitmq.New(queueExchange)
-				mq.RegisterReceiver(t)
-				err :=mq.Start()
-				if err != nil{
-
-					fmt.Println(err)
-				}
-				wg.Done()
-			}(&wg)
-		}
-		wg.Wait()
-		//fmt.Println("执行任务完成....")
-		time.Sleep(time.Microsecond*10)
-	}
 }
\ No newline at end of file