[agent-go] 新增线程池部分代码

.run/go build agent-go.run.xml

@@ -0,0 +1,13 @@
+<component name="ProjectRunConfigurationManager">
+    <configuration default="false" name="go build agent-go" type="GoApplicationRunConfiguration"
+                   factoryName="Go Application" nameIsGenerated="true">
+        <module name="ProjectOctopus"/>
+        <working_directory value="$PROJECT_DIR$/agent-go"/>
+        <parameters value="-version=dev"/>
+        <kind value="PACKAGE"/>
+        <package value="agent-go"/>
+        <directory value="$PROJECT_DIR$"/>
+        <filePath value="$PROJECT_DIR$/agent-go/main.go"/>
+        <method v="2"/>
+    </configuration>
+</component>

agent-go/g/global.go

@@ -2,6 +2,7 @@ package g
 
 import (
 	"agent-go/config"
+	"github.com/panjf2000/ants/v2"
 	"github.com/spf13/viper"
 )
 
@@ -9,6 +10,7 @@ type Global struct {
 	LOG             *Logger
 	NacosConfig     *viper.Viper
 	AgentServerInfo *config.AgentServerInfo
+	P               *ants.Pool
 }
 
 const (
@@ -21,11 +23,19 @@ const (
 
 var logger, _ = NewLogger()
 
+var pool, _ = ants.NewPool(100, ants.WithNonblocking(true), ants.WithLogger(logger))
+
 var G = NewGlobal(
 	logger,
+	pool,
 )
 
 // NewGlobal NewGlobal构造函数返回一个新的Global实例，其中包含指定的Logger。
-func NewGlobal(logger *Logger) *Global {
+func NewGlobal(logger *Logger, pool *ants.Pool) *Global {
-	return &Global{LOG: logger}
+	return &Global{
+		LOG:             logger,
+		NacosConfig:     nil,
+		AgentServerInfo: nil,
+		P:               pool,
+	}
 }

agent-go/g/logger.go

@@ -1,6 +1,7 @@
 package g
 
 import (
+	"fmt"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
@@ -34,16 +35,30 @@ func NewLogger() (*Logger, error) {
 	return &Logger{logger}, nil
 }
 
+func (l *Logger) Printf(msg string, args ...interface{}) {
+
+	l.Logger.Info(fmt.Sprintf("%s ==> %v", msg, args))
+}
+
 // Debug logs a debug message.
 func (l *Logger) Debug(msg string, fields ...zap.Field) {
 	l.Logger.Debug(msg, fields...)
 }
 
+func (l *Logger) DebugF(msg string, args ...interface{}) {
+	l.Logger.Debug(fmt.Sprintf(msg, args...))
+}
+
 // Info logs an info message.
 func (l *Logger) Info(msg string, fields ...zap.Field) {
 	l.Logger.Info(msg, fields...)
 }
 
+// InfoF logs an info message with format
+func (l *Logger) InfoF(msg string, args ...interface{}) {
+	l.Logger.Info(fmt.Sprintf(msg, args...))
+}
+
 // Warn logs a warning message.
 func (l *Logger) Warn(msg string, fields ...zap.Field) {
 	l.Logger.Warn(msg, fields...)

agent-go/go.mod

@@ -4,6 +4,7 @@ go 1.18
 
 require (
 	github.com/nacos-group/nacos-sdk-go/v2 v2.2.0
+	github.com/panjf2000/ants/v2 v2.7.2
 	github.com/spf13/viper v1.15.0
 	github.com/streadway/amqp v1.0.0
 	go.uber.org/zap v1.24.0

agent-go/go.sum

@@ -206,6 +206,8 @@ github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRW
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/nacos-group/nacos-sdk-go/v2 v2.2.0 h1:cAidHbdJjqUJOI1JXDlVe5eAL8pgZ/ORHoRj/ZXn/3o=
 github.com/nacos-group/nacos-sdk-go/v2 v2.2.0/go.mod h1:ys/1adWeKXXzbNWfRNbaFlX/t6HVLWdpsNDvmoWTw0g=
+github.com/panjf2000/ants/v2 v2.7.2 h1:2NUt9BaZFO5kQzrieOmK/wdb/tQ/K+QHaxN8sOgD63U=
+github.com/panjf2000/ants/v2 v2.7.2/go.mod h1:KIBmYG9QQX5U2qzFP/yQJaq/nSb6rahS9iEHkrCMgM8=
 github.com/pelletier/go-toml/v2 v2.0.6 h1:nrzqCb7j9cDFj2coyLNLaZuJTLjWjlaz6nvTvIwycIU=
 github.com/pelletier/go-toml/v2 v2.0.6/go.mod h1:eumQOmlWiOPt5WriQQqoM5y18pDHwha2N+QD+EUNTek=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

agent-go/rabbitmq/OctopusMsgHandler.go

@@ -8,6 +8,8 @@ import (
 	"fmt"
 )
 
+var P = g.G.P
+
 func HandleOMsg(initOMsgFromServer *config.OctopusMessage) {
 
 	agentTopicName := initOMsgFromServer.Result.(string)
@@ -65,11 +67,17 @@ func doHandleOctopusMessage(octopusMessage *config.OctopusMessage) {
 	case g.InitOmType:
 		go func() {}()
 	case g.ExecOmType:
-		go executorOMHandler(octopusMessage)
+		P.Submit(func() {
+			executorOMHandler(octopusMessage)
+		})
 	case g.StatusOmType:
-		go statusOMHandler(octopusMessage)
+		P.Submit(func() {
+			statusOMHandler(octopusMessage)
+		})
 	default:
-		go blackHoleOMHandler(octopusMessage)
+		P.Submit(func() {
+			blackHoleOMHandler(octopusMessage)
+		})
 	}
 
 }