feat: 给消息处理通道增加容量，有利于处理大量数据

koch3092 · Apr 9, 2023 · efc9888 · efc9888
1 parent ac7074e
commit efc9888
Show file tree

Hide file tree

Showing 4 changed files with 8 additions and 4 deletions.
diff --git a/initialize/amqp_manager.go b/initialize/amqp_manager.go
@@ -54,9 +54,11 @@ func (am *AmqpManager) StartReceiveMessage(ctx context.Context, sdRcvMsg chan<-
 			aRcvMsg <- message
 			err := message.Accept()
 			if err != nil {
+				am.Logger.Error(fmt.Sprintf("amqp receive data accept error:%s", err))
 				childDone()
 				return
 			}
+			am.Logger.Info("amqp receive data accept ok.")
 		} else {
 			fmt.Println("amqp receive data error:", err)
 

diff --git a/initialize/message_handler.go b/initialize/message_handler.go
@@ -32,6 +32,7 @@ ProcessMessage:
 	for {
 		select {
 		case message := <-rcvMessage:
+			h.Logger.Debug(fmt.Sprintf("Save data to be processed: %d", len(rcvMessage)))
 			var dataType model.DataType
 			errJson := json.Unmarshal(message.GetData(), &dataType)
 			if errJson != nil {
@@ -135,6 +136,7 @@ ProcessMessage:
 	for {
 		select {
 		case message := <-rcvMessage:
+			h.Logger.Debug(fmt.Sprintf("Measurement data to be processed: %d", len(rcvMessage)))
 			// 读取公共数据
 			var tdMetricBase *model.TdMetricBase
 			errTd := json.Unmarshal(message.GetData(), &tdMetricBase)

diff --git a/initialize/orm_tdengine.go b/initialize/orm_tdengine.go
@@ -8,7 +8,7 @@ import (
 func InitTdengine() (*sql.DB, error) {
 	m := global.CONFIG.TDengine
 	dsn := m.Dsn()
-	global.Logger.Debug("Open TDengin dsn: " + dsn)
+	global.Logger.Debug("Open TDengine dsn: " + dsn)
 	tdengine, err := sql.Open("taosSql", dsn)
 	if err != nil {
 		global.Logger.Debug("Open TDengine failed: " + err.Error())

diff --git a/main.go b/main.go
@@ -59,9 +59,9 @@ func main() {
 	ctx := context.Background()
 
 	// 用于转发消息的Channel
-	sdRcvMsg := make(chan *amqp.Message) // 保存原始数据用
-	mRcvMsg := make(chan *amqp.Message)  // 保存Measurement数据用
-	aRcvMsg := make(chan *amqp.Message)  // 告警用
+	sdRcvMsg := make(chan *amqp.Message, 200) // 保存原始数据用
+	mRcvMsg := make(chan *amqp.Message, 200)  // 保存Measurement数据用
+	aRcvMsg := make(chan *amqp.Message, 200)  // 告警用
 
 	// 阿里云AMQP凭证对象
 	aliCred := global.CONFIG.AliAmqpCred