优化短线后cpu飙升问题

app/http/tcppool/single.go

@@ -88,14 +88,17 @@ func (t *TcpHelper) SendMsg(msg []byte) error {
 					return nil
 				}
 			} else {
-				fmt.Println("结果：recvStr:", recvStr)
-				if recvStr == "5" {
-					fmt.Println("客户端繁忙")
-					atomic.StoreInt32(t.Full, 1)
-				} else if recvStr == "2" {
-					fmt.Println("客户端空闲")
-					atomic.StoreInt32(t.Full, 0)
+				if len(recvStr) > 0 {
+					fmt.Println("结果：recvStr:", recvStr)
+					if recvStr == "5" {
+						fmt.Println("客户端繁忙")
+						atomic.StoreInt32(t.Full, 1)
+					} else if recvStr == "2" {
+						fmt.Println("客户端空闲")
+						atomic.StoreInt32(t.Full, 0)
+					}
 				}
+
 			}
 
 		}

app/utils/mq/kafka_v2.go

@@ -14,6 +14,11 @@ import (
 	"time"
 )
 
+var (
+	now     int32 = 0
+	workNum       = &now
+)
+
 type KafkaV2Mq struct {
 }
 
@@ -90,7 +95,7 @@ func (kk KafkaV2Mq) Consume(name string, hand interface{}, ci int) {
 		go func() {
 			for {
 				if atomic.LoadInt32(tcppool.TcpFactory.Full) == 1 {
-					//utils.Log(nil, "对列阻塞")
+					utils.Log(nil, "对列阻塞")
 					time.Sleep(100 * time.Millisecond)
 					continue
 				} else {
@@ -104,7 +109,7 @@ func (kk KafkaV2Mq) Consume(name string, hand interface{}, ci int) {
 					var mqsg = entities.MqMessage{}
 					mqsg.Key = string(msg.Key)
 					mqsg.Property = make(map[string]interface{})
-					if false {
+					if len(msg.Value) > 0 {
 						if msg.Headers != nil {
 							for _, v := range msg.Headers {
 								if v.Key == "property" {
@@ -129,9 +134,10 @@ func (kk KafkaV2Mq) Consume(name string, hand interface{}, ci int) {
 					if err == nil {
 						//手动提交编译量
 						kk.commitOffset(consumer, msg.TopicPartition)
+						atomic.AddInt32(workNum, 1)
 						tcppool.OrderMap.Delete(string(msg.Key))
 						end = time.Now()
-						utils.Log(nil, "消费耗时", end.Sub(start))
+						utils.Log(nil, "消费耗时", end.Sub(start), "消息数", *workNum)
 
 					}
 				} else {