新增etcd认证配置

2.废弃SetGoRoutineNum接口
3.释放Module优化

cluster/etcddiscovery.go

@@ -13,10 +13,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"go.uber.org/zap"
 	"path"
 	"strings"
 	"sync/atomic"
+	"io/ioutil"
+	"crypto/x509"
+	"crypto/tls"
 )
 
 const originDir = "/origin"
@@ -40,11 +42,16 @@ type EtcdDiscoveryService struct {
 	mapDiscoveryNodeId map[string]map[string]struct{} //map[networkName]map[nodeId]
 }
 
+var  etcdDiscovery *EtcdDiscoveryService
 func getEtcdDiscovery() IServiceDiscovery {
-	etcdDiscovery := &EtcdDiscoveryService{}
+	if etcdDiscovery == nil {
+		etcdDiscovery = &EtcdDiscoveryService{}
+	}
+
 	return etcdDiscovery
 }
 
+
 func (ed *EtcdDiscoveryService) InitDiscovery(localNodeId string, funDelNode FunDelNode, funSetNode FunSetNode) error {
 	ed.localNodeId = localNodeId
 
@@ -87,15 +94,44 @@ func (ed *EtcdDiscoveryService) OnInit() error {
 	}
 
 	for i := 0; i < len(etcdDiscoveryCfg.EtcdList); i++ {
-		client, cerr := clientv3.New(clientv3.Config{
+		var client *clientv3.Client
+		var tlsConfig *tls.Config
+
+		if etcdDiscoveryCfg.EtcdList[i].Cert != "" {
+			// load cert
+			cert, cerr := tls.LoadX509KeyPair(etcdDiscoveryCfg.EtcdList[i].Cert, etcdDiscoveryCfg.EtcdList[i].CertKey)
+			if cerr != nil {
+				log.Error("load cert error", log.ErrorField("err", cerr))
+				return cerr
+			}
+
+			// load root ca
+			caData, cerr := ioutil.ReadFile(etcdDiscoveryCfg.EtcdList[i].Ca)
+			if cerr != nil {
+				log.Error("load root ca error", log.ErrorField("err", cerr))
+				return cerr
+			}
+			pool := x509.NewCertPool()
+			pool.AppendCertsFromPEM(caData)
+			tlsConfig = &tls.Config{
+				Certificates: []tls.Certificate{cert},
+				RootCAs:      pool,
+			}
+		}
+
+		client, err = clientv3.New(clientv3.Config{
 			Endpoints:   etcdDiscoveryCfg.EtcdList[i].Endpoints,
 			DialTimeout: etcdDiscoveryCfg.DialTimeoutMillisecond,
-			Logger:      zap.NewNop(),
+			Username: etcdDiscoveryCfg.EtcdList[i].UserName,
+			Password: etcdDiscoveryCfg.EtcdList[i].Password,
+			Logger:      log.GetLogger().Logger,
+			TLS: tlsConfig,
 		})
 
-		if cerr != nil {
-			log.Error("etcd discovery init fail", log.ErrorField("err", cerr))
-			return cerr
+
+		if err != nil {
+			log.Error("etcd discovery init fail", log.ErrorField("err", err))
+			return err
 		}
 
 		ctx, _ := context.WithTimeout(context.Background(), time.Second*3)

cluster/parsecfg.go

@@ -15,9 +15,15 @@ import (
 
 var json = jsoniter.ConfigCompatibleWithStandardLibrary
 
+
 type EtcdList struct {
 	NetworkName []string
 	Endpoints   []string
+	UserName string
+	Password string
+	Cert string
+	CertKey string
+	Ca string
 }
 
 type EtcdDiscovery struct {
@@ -417,13 +423,12 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
 	return nil
 }
 
-func (cls *Cluster) parseLocalCfg() {
+func (cls *Cluster) parseLocalCfg() error{
 	rpcInfo := NodeRpcInfo{}
 	rpcInfo.nodeInfo = cls.localNodeInfo
 	rpcInfo.client = rpc.NewLClient(rpcInfo.nodeInfo.NodeId, &cls.callSet)
 
 	cls.mapRpc[cls.localNodeInfo.NodeId] = &rpcInfo
-
 	for _, serviceName := range cls.localNodeInfo.ServiceList {
 		splitServiceName := strings.Split(serviceName, ":")
 		if len(splitServiceName) == 2 {
@@ -440,8 +445,13 @@ func (cls *Cluster) parseLocalCfg() {
 			cls.mapServiceNode[serviceName] = make(map[string]struct{})
 		}
 
+		if _,ok:=cls.mapServiceNode[serviceName][cls.localNodeInfo.NodeId];ok {
+			return fmt.Errorf("duplicate service %s is configured in node %s", serviceName, cls.localNodeInfo.NodeId)
+		}
 		cls.mapServiceNode[serviceName][cls.localNodeInfo.NodeId] = struct{}{}
 	}
+
+	return nil
 }
 
 func (cls *Cluster) IsNatsMode() bool {
@@ -474,8 +484,7 @@ func (cls *Cluster) InitCfg(localNodeId string) error {
 	}
 
 	//本地配置服务加到全局map信息中
-	cls.parseLocalCfg()
-	return nil
+	return cls.parseLocalCfg()
 }
 
 func (cls *Cluster) IsConfigService(serviceName string) bool {

log/log.go

@@ -18,14 +18,16 @@ type Logger struct {
 	OpenConsole   *bool
 	LogPath       string
 	FileName      string
+	Skip          int
 	LogLevel      zapcore.Level
 	Encoder       zapcore.Encoder
 	LogConfig     *lumberjack.Logger
-	sugaredLogger *zap.SugaredLogger
+	SugaredLogger *zap.SugaredLogger
+	CoreList      []zapcore.Core
 }
 
 func SetLogger(logger *Logger) {
-	if logger != nil && isSetLogger == false {
+	if logger != nil {
 		gLogger = logger
 		isSetLogger = true
 	}
@@ -39,6 +41,10 @@ func (logger *Logger) SetEncoder(encoder zapcore.Encoder) {
 	logger.Encoder = encoder
 }
 
+func (logger *Logger) SetSkip(skip int) {
+	logger.Skip = skip
+}
+
 func GetJsonEncoder() zapcore.Encoder {
 	encoderConfig := zap.NewProductionEncoderConfig()
 	encoderConfig.EncodeLevel = zapcore.CapitalLevelEncoder
@@ -90,22 +96,27 @@ func (logger *Logger) Enabled(zapcore.Level) bool {
 }
 
 func (logger *Logger) Init() {
-	var coreList []zapcore.Core
+	if isSetLogger {
+		return
+	}
 
+	var coreList []zapcore.Core
 	if logger.OpenConsole == nil || *logger.OpenConsole {
 		core := zapcore.NewCore(logger.Encoder, zapcore.AddSync(os.Stdout), logger.LogLevel)
 		coreList = append(coreList, core)
 	}
 
-	if logger.LogPath != "" {
-		writeSyncer := zapcore.AddSync(logger.LogConfig)
-		core := zapcore.NewCore(logger.Encoder, writeSyncer, logger.LogLevel)
+	if logger.CoreList != nil {
+		coreList = append(coreList, logger.CoreList...)
+	}else if logger.LogPath != "" {
+		WriteSyncer := zapcore.AddSync(logger.LogConfig)
+		core := zapcore.NewCore(logger.Encoder, WriteSyncer, logger.LogLevel)
 		coreList = append(coreList, core)
 	}
 
 	core := zapcore.NewTee(coreList...)
-	logger.Logger = zap.New(core, zap.AddCaller(), zap.AddStacktrace(logger), zap.AddCallerSkip(1))
-	logger.sugaredLogger = logger.Logger.Sugar()
+	logger.Logger = zap.New(core, zap.AddCaller(), zap.AddStacktrace(logger), zap.AddCallerSkip(1+logger.Skip))
+	logger.SugaredLogger = logger.Logger.Sugar()
 }
 
 func (logger *Logger) Debug(msg string, fields ...zap.Field) {
@@ -164,85 +175,85 @@ func Fatal(msg string, fields ...zap.Field) {
 	gLogger.stack = false
 }
 
-func Debugf(msg string, args ...any) {
-	gLogger.sugaredLogger.Debugf(msg, args...)
+func Debugf(template string, args ...any) {
+	gLogger.SugaredLogger.Debugf(template, args...)
 }
 
-func Infof(msg string, args ...any) {
-	gLogger.sugaredLogger.Infof(msg, args...)
+func Infof(template string, args ...any) {
+	gLogger.SugaredLogger.Infof(template, args...)
 }
 
-func Warnf(msg string, args ...any) {
-	gLogger.sugaredLogger.Warnf(msg, args...)
+func Warnf(template string, args ...any) {
+	gLogger.SugaredLogger.Warnf(template, args...)
 }
 
-func Errorf(msg string, args ...any) {
-	gLogger.sugaredLogger.Errorf(msg, args...)
+func Errorf(template string, args ...any) {
+	gLogger.SugaredLogger.Errorf(template, args...)
 }
 
-func StackErrorf(msg string, args ...any) {
+func StackErrorf(template string, args ...any) {
 	gLogger.stack = true
-	gLogger.sugaredLogger.Errorf(msg, args...)
+	gLogger.SugaredLogger.Errorf(template, args...)
 	gLogger.stack = false
 }
 
-func Fatalf(msg string, args ...any) {
-	gLogger.sugaredLogger.Fatalf(msg, args...)
+func Fatalf(template string, args ...any) {
+	gLogger.SugaredLogger.Fatalf(template, args...)
 }
 
 func (logger *Logger) SDebug(args ...interface{}) {
-	logger.sugaredLogger.Debugln(args...)
+	logger.SugaredLogger.Debugln(args...)
 }
 
 func (logger *Logger) SInfo(args ...interface{}) {
-	logger.sugaredLogger.Infoln(args...)
+	logger.SugaredLogger.Infoln(args...)
 }
 
 func (logger *Logger) SWarn(args ...interface{}) {
-	logger.sugaredLogger.Warnln(args...)
+	logger.SugaredLogger.Warnln(args...)
 }
 
 func (logger *Logger) SError(args ...interface{}) {
-	logger.sugaredLogger.Errorln(args...)
+	logger.SugaredLogger.Errorln(args...)
 }
 
 func (logger *Logger) SStackError(args ...interface{}) {
 	gLogger.stack = true
-	logger.sugaredLogger.Errorln(args...)
+	logger.SugaredLogger.Errorln(args...)
 	gLogger.stack = false
 }
 
 func (logger *Logger) SFatal(args ...interface{}) {
 	gLogger.stack = true
-	logger.sugaredLogger.Fatalln(args...)
+	logger.SugaredLogger.Fatalln(args...)
 	gLogger.stack = false
 }
 
 func SDebug(args ...interface{}) {
-	gLogger.sugaredLogger.Debugln(args...)
+	gLogger.SugaredLogger.Debugln(args...)
 }
 
 func SInfo(args ...interface{}) {
-	gLogger.sugaredLogger.Infoln(args...)
+	gLogger.SugaredLogger.Infoln(args...)
 }
 
 func SWarn(args ...interface{}) {
-	gLogger.sugaredLogger.Warnln(args...)
+	gLogger.SugaredLogger.Warnln(args...)
 }
 
 func SError(args ...interface{}) {
-	gLogger.sugaredLogger.Errorln(args...)
+	gLogger.SugaredLogger.Errorln(args...)
 }
 
 func SStackError(args ...interface{}) {
 	gLogger.stack = true
-	gLogger.sugaredLogger.Errorln(args...)
+	gLogger.SugaredLogger.Errorln(args...)
 	gLogger.stack = false
 }
 
 func SFatal(args ...interface{}) {
 	gLogger.stack = true
-	gLogger.sugaredLogger.Fatalln(args...)
+	gLogger.SugaredLogger.Fatalln(args...)
 	gLogger.stack = false
 }
 

node/node.go

@@ -230,7 +230,6 @@ func initLog() error {
 	localNodeInfo := cluster.GetCluster().GetLocalNodeInfo()
 	fileName := fmt.Sprintf("%s.log", localNodeInfo.NodeId)
 	logger.FileName = fileName
-	filepath.Join()
 	logger.LogConfig.Filename = filepath.Join(logger.LogPath, logger.FileName)
 
 	logger.Init()
@@ -330,13 +329,13 @@ func startNode(args interface{}) error {
 		myName, mErr := sysprocess.GetMyProcessName()
 		//当前进程名获取失败，不应该发生
 		if mErr != nil {
-			log.Info("get my process's name is error", log.ErrorField("err", mErr))
+			log.Error("get my process's name is error", log.ErrorField("err", mErr))
 			os.Exit(-1)
 		}
 
 		//进程id存在，而且进程名也相同，被认为是当前进程重复运行
 		if cErr == nil && name == myName {
-			log.Info("repeat runs are not allowed", log.String("nodeId", strNodeId), log.Int("processId", processId))
+			log.Error("repeat runs are not allowed", log.String("nodeId", strNodeId), log.Int("processId", processId))
 			os.Exit(-1)
 		}
 		break
@@ -354,10 +353,13 @@ func startNode(args interface{}) error {
 	service.Start()
 
 	//5.运行集群
-	cluster.GetCluster().Start()
+	err := cluster.GetCluster().Start()
+	if err != nil {
+		log.Error(err.Error())
+		os.Exit(-1)
+	}
 
 	//6.监听程序退出信号&性能报告
-
 	var pProfilerTicker *time.Ticker = &time.Ticker{}
 	if profilerInterval > 0 {
 		pProfilerTicker = time.NewTicker(profilerInterval)
@@ -484,9 +486,9 @@ func setLogPath(args interface{}) error {
 	}
 
 	if err != nil {
-		err = os.Mkdir(log.GetLogger().LogPath, os.ModePerm)
+		err = os.MkdirAll(logPath, os.ModePerm)
 		if err != nil {
-			return errors.New("Cannot create dir " + log.GetLogger().LogPath)
+			return errors.New("Cannot create dir " + logPath)
 		}
 	}
 

rpc/client.go

@@ -32,7 +32,7 @@ type IRealClient interface {
 	SetConn(conn *network.NetConn)
 	Close(waitDone bool)
 
-	AsyncCall(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}, cancelable bool) (CancelRpc, error)
+	AsyncCall(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error)
 	Go(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, noReply bool, serviceMethod string, args interface{}, reply interface{}) *Call
 	RawGo(NodeId string, timeout time.Duration, rpcHandler IRpcHandler, processor IRpcProcessor, noReply bool, rpcMethodId uint32, serviceMethod string, rawArgs []byte, reply interface{}) *Call
 	IsConnected() bool
@@ -211,7 +211,7 @@ func (client *Client) rawGo(nodeId string, w IWriter, timeout time.Duration, rpc
 	return call
 }
 
-func (client *Client) asyncCall(nodeId string, w IWriter, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}, cancelable bool) (CancelRpc, error) {
+func (client *Client) asyncCall(nodeId string, w IWriter, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error) {
 	processorType, processor := GetProcessorType(args)
 	InParam, herr := processor.Marshal(args)
 	if herr != nil {
@@ -264,10 +264,7 @@ func (client *Client) asyncCall(nodeId string, w IWriter, timeout time.Duration,
 		return emptyCancelRpc, err
 	}
 
-	if cancelable {
-		rpcCancel := RpcCancel{CallSeq: seq, Cli: client}
-		return rpcCancel.CancelRpc, nil
-	}
 
-	return emptyCancelRpc, nil
+	rpcCancel := RpcCancel{CallSeq: seq, Cli: client}
+	return rpcCancel.CancelRpc, nil
 }

rpc/lclient.go

@@ -90,7 +90,7 @@ func (lc *LClient) RawGo(nodeId string, timeout time.Duration, rpcHandler IRpcHa
 	return pLocalRpcServer.selfNodeRpcHandlerGo(timeout, processor, lc.selfClient, true, serviceName, rpcMethodId, serviceName, nil, nil, rawArgs)
 }
 
-func (lc *LClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, reply interface{}, cancelable bool) (CancelRpc, error) {
+func (lc *LClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, reply interface{}) (CancelRpc, error) {
 	pLocalRpcServer := rpcHandler.GetRpcServer()()
 
 	//判断是否是同一服务
@@ -109,7 +109,7 @@ func (lc *LClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IR
 	}
 
 	//其他的rpcHandler的处理器
-	cancelRpc, err := pLocalRpcServer.selfNodeRpcHandlerAsyncGo(timeout, lc.selfClient, rpcHandler, false, serviceName, serviceMethod, args, reply, callback, cancelable)
+	cancelRpc, err := pLocalRpcServer.selfNodeRpcHandlerAsyncGo(timeout, lc.selfClient, rpcHandler, false, serviceName, serviceMethod, args, reply, callback)
 	if err != nil {
 		callback.Call([]reflect.Value{reflect.ValueOf(reply), reflect.ValueOf(err)})
 	}
@@ -121,9 +121,6 @@ func NewLClient(localNodeId string, callSet *CallSet) *Client {
 	client := &Client{}
 	client.clientId = atomic.AddUint32(&clientSeq, 1)
 	client.targetNodeId = localNodeId
-	//client.maxCheckCallRpcCount = DefaultMaxCheckCallRpcCount
-	//client.callRpcTimeout = DefaultRpcTimeout
-
 	lClient := &LClient{}
 	lClient.selfClient = client
 	client.IRealClient = lClient

rpc/lserver.go

@@ -127,7 +127,7 @@ func (server *BaseServer) selfNodeRpcHandlerGo(timeout time.Duration, processor
 	return pCall
 }
 
-func (server *BaseServer) selfNodeRpcHandlerAsyncGo(timeout time.Duration, client *Client, callerRpcHandler IRpcHandler, noReply bool, handlerName string, serviceMethod string, args interface{}, reply interface{}, callback reflect.Value, cancelable bool) (CancelRpc, error) {
+func (server *BaseServer) selfNodeRpcHandlerAsyncGo(timeout time.Duration, client *Client, callerRpcHandler IRpcHandler, noReply bool, handlerName string, serviceMethod string, args interface{}, reply interface{}, callback reflect.Value) (CancelRpc, error) {
 	rpcHandler := server.rpcHandleFinder.FindRpcHandler(handlerName)
 	if rpcHandler == nil {
 		err := errors.New("service method " + serviceMethod + " not config!")

rpc/natsclient.go

@@ -63,8 +63,8 @@ func (nc *NatsClient) RawGo(nodeId string, timeout time.Duration, rpcHandler IRp
 	return nc.client.rawGo(nodeId, nc, timeout, rpcHandler, processor, noReply, rpcMethodId, serviceMethod, rawArgs, reply)
 }
 
-func (nc *NatsClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}, cancelable bool) (CancelRpc, error) {
-	cancelRpc, err := nc.client.asyncCall(nodeId, nc, timeout, rpcHandler, serviceMethod, callback, args, replyParam, cancelable)
+func (nc *NatsClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error) {
+	cancelRpc, err := nc.client.asyncCall(nodeId, nc, timeout, rpcHandler, serviceMethod, callback, args, replyParam)
 	if err != nil {
 		callback.Call([]reflect.Value{reflect.ValueOf(replyParam), reflect.ValueOf(err)})
 	}

rpc/rclient.go

@@ -62,8 +62,8 @@ func (rc *RClient) RawGo(nodeId string, timeout time.Duration, rpcHandler IRpcHa
 	return rc.selfClient.rawGo(nodeId, rc, timeout, rpcHandler, processor, noReply, rpcMethodId, serviceMethod, rawArgs, reply)
 }
 
-func (rc *RClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}, cancelable bool) (CancelRpc, error) {
-	cancelRpc, err := rc.selfClient.asyncCall(nodeId, rc, timeout, rpcHandler, serviceMethod, callback, args, replyParam, cancelable)
+func (rc *RClient) AsyncCall(nodeId string, timeout time.Duration, rpcHandler IRpcHandler, serviceMethod string, callback reflect.Value, args interface{}, replyParam interface{}) (CancelRpc, error) {
+	cancelRpc, err := rc.selfClient.asyncCall(nodeId, rc, timeout, rpcHandler, serviceMethod, callback, args, replyParam)
 	if err != nil {
 		callback.Call([]reflect.Value{reflect.ValueOf(replyParam), reflect.ValueOf(err)})
 	}

rpc/rpchandler.go

@@ -164,13 +164,6 @@ func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
 	//取出输入参数类型
 	var rpcMethodInfo RpcMethodInfo
 	typ := method.Type
-	if typ.NumOut() != 1 {
-		return fmt.Errorf("%s The number of returned arguments must be 1", method.Name)
-	}
-
-	if typ.Out(0).String() != "error" {
-		return fmt.Errorf("%s The return parameter must be of type error", method.Name)
-	}
 
 	if typ.NumIn() < 2 || typ.NumIn() > 4 {
 		return fmt.Errorf("%s Unsupported parameter format", method.Name)
@@ -183,6 +176,18 @@ func (handler *RpcHandler) suitableMethods(method reflect.Method) error {
 		rpcMethodInfo.hasResponder = true
 	}
 
+	if rpcMethodInfo.hasResponder && typ.NumOut() > 0  {
+		return fmt.Errorf("%s should not have return parameters", method.Name)
+	}
+
+	if !rpcMethodInfo.hasResponder && typ.NumOut() != 1  {
+		return fmt.Errorf("%s The number of returned arguments must be 1", method.Name)
+	}
+
+	if !rpcMethodInfo.hasResponder && typ.Out(0).String() != "error" {
+		return fmt.Errorf("%s The return parameter must be of type error", method.Name)
+	}
+
 	for i := parIdx; i < typ.NumIn(); i++ {
 		if handler.isExportedOrBuiltinType(typ.In(i)) == false {
 			return fmt.Errorf("%s Unsupported parameter types", method.Name)
@@ -307,9 +312,11 @@ func (handler *RpcHandler) HandlerRpcRequest(request *RpcRequest) {
 
 	requestHandle := request.requestHandle
 	returnValues := v.method.Func.Call(paramList)
-	errInter := returnValues[0].Interface()
-	if errInter != nil {
-		err = errInter.(error)
+	if len(returnValues) > 0 {
+		errInter := returnValues[0].Interface()
+		if errInter != nil {
+			err = errInter.(error)
+		}
 	}
 
 	if v.hasResponder == false && requestHandle != nil {
@@ -526,8 +533,7 @@ func (handler *RpcHandler) asyncCallRpc(timeout time.Duration, nodeId string, se
 	}
 
 	//2.rpcClient调用
-	//如果调用本结点服务
-	return pClientList[0].AsyncCall(pClientList[0].GetTargetNodeId(), timeout, handler.rpcHandler, serviceMethod, fVal, args, reply, false)
+	return pClientList[0].AsyncCall(pClientList[0].GetTargetNodeId(), timeout, handler.rpcHandler, serviceMethod, fVal, args, reply, )
 }
 
 func (handler *RpcHandler) GetName() string {

rpc/rpcnats.go

@@ -27,9 +27,6 @@ func  (rn *RpcNats) NewNatsClient(targetNodeId string,localNodeId string,callSet
 
 	client.clientId = atomic.AddUint32(&clientSeq, 1)
 	client.targetNodeId = targetNodeId
-	//client.maxCheckCallRpcCount = DefaultMaxCheckCallRpcCount
-	//client.callRpcTimeout = DefaultRpcTimeout
-
 	natsClient := &rn.NatsClient
 	natsClient.localNodeId = localNodeId
 	natsClient.client = &client

rpc/server.go

@@ -31,7 +31,7 @@ type IServer interface {
 
 	selfNodeRpcHandlerGo(timeout time.Duration, processor IRpcProcessor, client *Client, noReply bool, handlerName string, rpcMethodId uint32, serviceMethod string, args interface{}, reply interface{}, rawArgs []byte) *Call
 	myselfRpcHandlerGo(client *Client, handlerName string, serviceMethod string, args interface{}, callBack reflect.Value, reply interface{}) error
-	selfNodeRpcHandlerAsyncGo(timeout time.Duration, client *Client, callerRpcHandler IRpcHandler, noReply bool, handlerName string, serviceMethod string, args interface{}, reply interface{}, callback reflect.Value, cancelable bool) (CancelRpc, error)
+	selfNodeRpcHandlerAsyncGo(timeout time.Duration, client *Client, callerRpcHandler IRpcHandler, noReply bool, handlerName string, serviceMethod string, args interface{}, reply interface{}, callback reflect.Value) (CancelRpc, error)
 }
 
 type writeResponse func(processor IRpcProcessor, connTag string, serviceMethod string, seq uint64, reply interface{}, rpcError RpcError)

service/module.go

@@ -11,6 +11,7 @@ import (
 	"github.com/duanhf2012/origin/v2/log"
 	rpcHandle "github.com/duanhf2012/origin/v2/rpc"
 	"github.com/duanhf2012/origin/v2/util/timer"
+	"slices"
 )
 
 const InitModuleId = 1e9
@@ -46,7 +47,7 @@ type Module struct {
 	moduleName       string             //模块名称
 	parent           IModule            //父亲
 	self             IModule            //自己
-	child            map[uint32]IModule //孩子们
+	child            []IModule //孩子们
 	mapActiveTimer   map[timer.ITimer]struct{}
 	mapActiveIdTimer map[uint64]timer.ITimer
 	dispatcher       *timer.Dispatcher //timer
@@ -93,10 +94,7 @@ func (m *Module) AddModule(module IModule) (uint32, error) {
 		pAddModule.moduleId = m.NewModuleId()
 	}
 
-	if m.child == nil {
-		m.child = map[uint32]IModule{}
-	}
-	_, ok := m.child[module.GetModuleId()]
+	_,ok := m.ancestor.getBaseModule().(*Module).descendants[module.GetModuleId()]
 	if ok == true {
 		return 0, fmt.Errorf("exists module id %d", module.GetModuleId())
 	}
@@ -109,29 +107,33 @@ func (m *Module) AddModule(module IModule) (uint32, error) {
 	pAddModule.eventHandler = event.NewEventHandler()
 	pAddModule.eventHandler.Init(m.eventHandler.GetEventProcessor())
 	pAddModule.IConcurrent = m.IConcurrent
+
+	m.child = append(m.child,module)
+	m.ancestor.getBaseModule().(*Module).descendants[module.GetModuleId()] = module
+
 	err := module.OnInit()
 	if err != nil {
+		delete(m.ancestor.getBaseModule().(*Module).descendants, module.GetModuleId())
+		m.child = m.child[:len(m.child)-1]
+		log.Error("module OnInit error",log.String("ModuleName",module.GetModuleName()),log.ErrorField("err",err))
 		return 0, err
 	}
 
-	m.child[module.GetModuleId()] = module
-	m.ancestor.getBaseModule().(*Module).descendants[module.GetModuleId()] = module
-
 	log.Debug("Add module " + module.GetModuleName() + " completed")
 	return module.GetModuleId(), nil
 }
 
 func (m *Module) ReleaseModule(moduleId uint32) {
 	pModule := m.GetModule(moduleId).getBaseModule().(*Module)
+	pModule.self.OnRelease()
+	log.Debug("Release module " + pModule.GetModuleName())
 
-	//释放子孙
-	for id := range pModule.child {
-		m.ReleaseModule(id)
+	for i:=len(pModule.child)-1; i>=0; i-- {
+		m.ReleaseModule(pModule.child[i].GetModuleId())
 	}
 
-	pModule.self.OnRelease()
 	pModule.GetEventHandler().Destroy()
-	log.Debug("Release module " + pModule.GetModuleName())
+
 	for pTimer := range pModule.mapActiveTimer {
 		pTimer.Cancel()
 	}
@@ -140,7 +142,10 @@ func (m *Module) ReleaseModule(moduleId uint32) {
 		t.Cancel()
 	}
 
-	delete(m.child, moduleId)
+	m.child = slices.DeleteFunc(m.child, func(module IModule) bool {
+		return module.GetModuleId() == moduleId
+	})
+
 	delete(m.ancestor.getBaseModule().(*Module).descendants, moduleId)
 
 	//清理被删除的Module

service/service.go

@@ -266,8 +266,10 @@ func (s *Service) Release() {
 
 	if atomic.AddInt32(&s.isRelease, -1) == -1 {
 		s.self.OnRelease()
+		for i:=len(s.child)-1; i>=0; i-- {
+			s.ReleaseModule(s.child[i].GetModuleId())
+		}
 	}
-
 }
 
 func (s *Service) OnRelease() {
@@ -432,6 +434,7 @@ func (s *Service) SetEventChannelNum(num int) {
 	}
 }
 
+// Deprecated: replace it with the OpenConcurrent function
 func (s *Service) SetGoRoutineNum(goroutineNum int32) bool {
 	//已经开始状态不允许修改协程数量,打开性能分析器不允许开多线程
 	if s.startStatus == true || s.profiler != nil {

sysmodule/netmodule/wsmodule/WSModule.go

@@ -14,7 +14,7 @@ import (
 type WSModule struct {
 	service.Module
 
-	wsServer network.WSServer
+	WSServer network.WSServer
 
 	mapClientLocker sync.RWMutex
 	mapClient       map[string]*WSClient
@@ -57,16 +57,16 @@ func (ws *WSModule) OnInit() error {
 		return fmt.Errorf("please call the Init function correctly")
 	}
 
-	ws.wsServer.MaxConnNum = ws.wsCfg.MaxConnNum
-	ws.wsServer.PendingWriteNum = ws.wsCfg.PendingWriteNum
-	ws.wsServer.MaxMsgLen = ws.wsCfg.MaxMsgLen
-	ws.wsServer.Addr = ws.wsCfg.ListenAddr
+	ws.WSServer.MaxConnNum = ws.wsCfg.MaxConnNum
+	ws.WSServer.PendingWriteNum = ws.wsCfg.PendingWriteNum
+	ws.WSServer.MaxMsgLen = ws.wsCfg.MaxMsgLen
+	ws.WSServer.Addr = ws.wsCfg.ListenAddr
 
 	//3.设置解析处理器
 	ws.process.SetByteOrder(ws.wsCfg.LittleEndian)
 
-	ws.mapClient = make(map[string]*WSClient, ws.wsServer.MaxConnNum)
-	ws.wsServer.NewAgent = ws.NewWSClient
+	ws.mapClient = make(map[string]*WSClient, ws.WSServer.MaxConnNum)
+	ws.WSServer.NewAgent = ws.NewWSClient
 
 	//4.设置网络事件处理
 	ws.GetEventProcessor().RegEventReceiverFunc(event.Sys_Event_WebSocket, ws.GetEventHandler(), ws.wsEventHandler)
@@ -80,7 +80,7 @@ func (ws *WSModule) Init(wsCfg *WSCfg, process processor.IRawProcessor) {
 }
 
 func (ws *WSModule) Start() error {
-	return ws.wsServer.Start()
+	return ws.WSServer.Start()
 }
 
 func (ws *WSModule) wsEventHandler(ev event.IEvent) {
@@ -197,3 +197,7 @@ func (ws *WSModule) SendRawMsg(clientId string, msg []byte) error {
 	ws.mapClientLocker.Unlock()
 	return client.wsConn.WriteMsg(msg)
 }
+
+func (ws *WSModule) SetMessageType(messageType int) {
+	ws.WSServer.SetMessageType(messageType)
+}

util/smath/smath.go

@@ -32,10 +32,10 @@ func Abs[NumType typ.Signed | typ.Float](Num NumType) NumType {
 func AddSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, bool) {
 	ret := number1 + number2
 	if number2 > 0 && ret < number1 {
-		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 		return ret, false
 	} else if number2 < 0 && ret > number1 {
-		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 		return ret, false
 	}
 
@@ -45,10 +45,10 @@ func AddSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, boo
 func SubSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, bool) {
 	ret := number1 - number2
 	if number2 > 0 && ret > number1 {
-		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 		return ret, false
 	} else if number2 < 0 && ret < number1 {
-		log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+		log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 		return ret, false
 	}
 
@@ -65,7 +65,7 @@ func MulSafe[NumType typ.Number](number1 NumType, number2 NumType) (NumType, boo
 		return ret, true
 	}
 
-	log.Stack("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
+	log.SStackError("Calculation overflow", log.Any("number1", number1), log.Any("number2", number2))
 	return ret, true
 }