Ma/origin
1
0
Fork 0
mirror of https://github.com/duanhf2012/origin.git synced 2026-04-14 13:07:30 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

优化代码规范

Browse Source
This commit is contained in:
duanhf2012
2024-09-20 17:25:08 +08:00
parent 1cf071a444
commit 39116c4402
63 changed files with 1971 additions and 2043 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
cluster/cluster.go
View File

@@ -1,15 +1,15 @@
package cluster
import (
"errors"
"fmt"
"github.com/duanhf2012/origin/v2/event"
"github.com/duanhf2012/origin/v2/log"
"github.com/duanhf2012/origin/v2/rpc"
"github.com/duanhf2012/origin/v2/service"
"reflect"
"strings"
"sync"
"github.com/duanhf2012/origin/v2/event"
"errors"
"reflect"
)
var configDir = "./config/"
@@ -24,22 +24,22 @@ const (
)
type DiscoveryService struct {
MasterNodeId string //要筛选的主结点Id，如果不配置或者配置成0，表示针对所有的主结点
NetworkName string //如果是etcd，指定要筛选的网络名中的服务，不配置，表示所有的网络
ServiceList []string //只发现的服务列表
MasterNodeId string //要筛选的主结点Id，如果不配置或者配置成0，表示针对所有的主结点
NetworkName string //如果是etcd，指定要筛选的网络名中的服务，不配置，表示所有的网络
ServiceList []string //只发现的服务列表
}
type NodeInfo struct {
NodeId string
Private bool
ListenAddr string
MaxRpcParamLen uint32 //最大Rpc参数长度
CompressBytesLen int //超过字节进行压缩的长度
ServiceList []string //所有的有序服务列表
PublicServiceList []string //对外公开的服务列表
MaxRpcParamLen uint32 //最大Rpc参数长度
CompressBytesLen int //超过字节进行压缩的长度
ServiceList []string //所有的有序服务列表
PublicServiceList []string //对外公开的服务列表
DiscoveryService []DiscoveryService //筛选发现的服务，如果不配置，不进行筛选
status NodeStatus
Retire bool
Retire bool
NetworkName string
}
@@ -52,22 +52,22 @@ type NodeRpcInfo struct {
var cluster Cluster
type Cluster struct {
localNodeInfo NodeInfo //本结点配置信息
localNodeInfo NodeInfo //本结点配置信息
discoveryInfo DiscoveryInfo //服务发现配置
rpcMode RpcMode
globalCfg interface{} //全局配置
rpcMode RpcMode
globalCfg interface{} //全局配置
localServiceCfg map[string]interface{} //map[serviceName]配置数据*
serviceDiscovery IServiceDiscovery //服务发现接口
locker sync.RWMutex //结点与服务关系保护锁
mapRpc map[string]*NodeRpcInfo //nodeId
mapServiceNode map[string]map[string]struct{} //map[serviceName]map[NodeId]
locker sync.RWMutex //结点与服务关系保护锁
mapRpc map[string]*NodeRpcInfo //nodeId
mapServiceNode map[string]map[string]struct{} //map[serviceName]map[NodeId]
mapTemplateServiceNode map[string]map[string]struct{} //map[templateServiceName]map[serviceName]nodeId
callSet rpc.CallSet
rpcNats rpc.RpcNats
callSet rpc.CallSet
rpcNats rpc.RpcNats
rpcServer rpc.IServer
rpcEventLocker sync.RWMutex //Rpc事件监听保护锁
@@ -86,7 +86,7 @@ func SetServiceDiscovery(serviceDiscovery IServiceDiscovery) {
cluster.serviceDiscovery = serviceDiscovery
}
func (cls *Cluster) Start() error{
func (cls *Cluster) Start() error {
return cls.rpcServer.Start()
}
@@ -97,7 +97,7 @@ func (cls *Cluster) Stop() {
func (cls *Cluster) DiscardNode(nodeId string) {
cls.locker.Lock()
nodeInfo, ok := cls.mapRpc[nodeId]
bDel := (ok == true) && nodeInfo.nodeInfo.status == Discard
bDel := (ok == true) && nodeInfo.nodeInfo.status == Discard
cls.locker.Unlock()
if bDel {
@@ -113,22 +113,22 @@ func (cls *Cluster) DelNode(nodeId string) {
cls.locker.Lock()
defer cls.locker.Unlock()
rpc, ok := cls.mapRpc[nodeId]
nodeRpc, ok := cls.mapRpc[nodeId]
if ok == false {
return
}
cls.TriggerDiscoveryEvent(false,nodeId,rpc.nodeInfo.ServiceList)
for _, serviceName := range rpc.nodeInfo.ServiceList {
cls.TriggerDiscoveryEvent(false, nodeId, nodeRpc.nodeInfo.ServiceList)
for _, serviceName := range nodeRpc.nodeInfo.ServiceList {
cls.delServiceNode(serviceName, nodeId)
}
delete(cls.mapRpc, nodeId)
if ok == true {
rpc.client.Close(false)
nodeRpc.client.Close(false)
}
log.Info("remove node ",log.String("NodeId", rpc.nodeInfo.NodeId),log.String("ListenAddr", rpc.nodeInfo.ListenAddr))
log.Info("remove node ", log.String("NodeId", nodeRpc.nodeInfo.NodeId), log.String("ListenAddr", nodeRpc.nodeInfo.ListenAddr))
}
func (cls *Cluster) serviceDiscoveryDelNode(nodeId string) {
@@ -141,16 +141,16 @@ func (cls *Cluster) delServiceNode(serviceName string, nodeId string) {
}
//处理模板服务
splitServiceName := strings.Split(serviceName,":")
splitServiceName := strings.Split(serviceName, ":")
if len(splitServiceName) == 2 {
serviceName = splitServiceName[0]
templateServiceName := splitServiceName[1]
mapService := cls.mapTemplateServiceNode[templateServiceName]
delete(mapService,serviceName)
delete(mapService, serviceName)
if len(cls.mapTemplateServiceNode[templateServiceName]) == 0 {
delete(cls.mapTemplateServiceNode,templateServiceName)
delete(cls.mapTemplateServiceNode, templateServiceName)
}
}
@@ -178,7 +178,7 @@ func (cls *Cluster) serviceDiscoverySetNodeInfo(nodeInfo *NodeInfo) {
}
}
cluster.TriggerDiscoveryEvent(true,nodeInfo.NodeId,nodeInfo.PublicServiceList)
cluster.TriggerDiscoveryEvent(true, nodeInfo.NodeId, nodeInfo.PublicServiceList)
//再重新组装
mapDuplicate := map[string]interface{}{} //预防重复数据
for _, serviceName := range nodeInfo.PublicServiceList {
@@ -190,13 +190,13 @@ func (cls *Cluster) serviceDiscoverySetNodeInfo(nodeInfo *NodeInfo) {
mapDuplicate[serviceName] = nil
//如果是模板服务，则记录模板关系
splitServiceName := strings.Split(serviceName,":")
splitServiceName := strings.Split(serviceName, ":")
if len(splitServiceName) == 2 {
serviceName = splitServiceName[0]
templateServiceName := splitServiceName[1]
//记录模板
if _, ok = cls.mapTemplateServiceNode[templateServiceName]; ok == false {
cls.mapTemplateServiceNode[templateServiceName]=map[string]struct{}{}
cls.mapTemplateServiceNode[templateServiceName] = map[string]struct{}{}
}
cls.mapTemplateServiceNode[templateServiceName][serviceName] = struct{}{}
}
@@ -208,7 +208,7 @@ func (cls *Cluster) serviceDiscoverySetNodeInfo(nodeInfo *NodeInfo) {
}
if lastNodeInfo != nil {
log.Info("Discovery nodeId",log.String("NodeId", nodeInfo.NodeId),log.Any("services:", nodeInfo.PublicServiceList),log.Bool("Retire",nodeInfo.Retire))
log.Info("Discovery nodeId", log.String("NodeId", nodeInfo.NodeId), log.Any("services:", nodeInfo.PublicServiceList), log.Bool("Retire", nodeInfo.Retire))
lastNodeInfo.nodeInfo = *nodeInfo
return
}
@@ -218,19 +218,18 @@ func (cls *Cluster) serviceDiscoverySetNodeInfo(nodeInfo *NodeInfo) {
rpcInfo.nodeInfo = *nodeInfo
if cls.IsNatsMode() {
rpcInfo.client = cls.rpcNats.NewNatsClient(nodeInfo.NodeId, cls.GetLocalNodeInfo().NodeId,&cls.callSet,cls.NotifyAllService)
}else{
rpcInfo.client =rpc.NewRClient(nodeInfo.NodeId, nodeInfo.ListenAddr, nodeInfo.MaxRpcParamLen,cls.localNodeInfo.CompressBytesLen,&cls.callSet,cls.NotifyAllService)
rpcInfo.client = cls.rpcNats.NewNatsClient(nodeInfo.NodeId, cls.GetLocalNodeInfo().NodeId, &cls.callSet, cls.NotifyAllService)
} else {
rpcInfo.client = rpc.NewRClient(nodeInfo.NodeId, nodeInfo.ListenAddr, nodeInfo.MaxRpcParamLen, cls.localNodeInfo.CompressBytesLen, &cls.callSet, cls.NotifyAllService)
}
cls.mapRpc[nodeInfo.NodeId] = &rpcInfo
if cls.IsNatsMode() == true || cls.discoveryInfo.discoveryType!=OriginType {
log.Info("Discovery nodeId and new rpc client",log.String("NodeId", nodeInfo.NodeId),log.Any("services:", nodeInfo.PublicServiceList),log.Bool("Retire",nodeInfo.Retire))
}else{
log.Info("Discovery nodeId and new rpc client",log.String("NodeId", nodeInfo.NodeId),log.Any("services:", nodeInfo.PublicServiceList),log.Bool("Retire",nodeInfo.Retire),log.String("nodeListenAddr",nodeInfo.ListenAddr))
if cls.IsNatsMode() == true || cls.discoveryInfo.discoveryType != OriginType {
log.Info("Discovery nodeId and new rpc client", log.String("NodeId", nodeInfo.NodeId), log.Any("services:", nodeInfo.PublicServiceList), log.Bool("Retire", nodeInfo.Retire))
} else {
log.Info("Discovery nodeId and new rpc client", log.String("NodeId", nodeInfo.NodeId), log.Any("services:", nodeInfo.PublicServiceList), log.Bool("Retire", nodeInfo.Retire), log.String("nodeListenAddr", nodeInfo.ListenAddr))
}
}
func (cls *Cluster) Init(localNodeId string, setupServiceFun SetupServiceFun) error {
//1.初始化配置
err := cls.InitCfg(localNodeId)
@@ -240,18 +239,18 @@ func (cls *Cluster) Init(localNodeId string, setupServiceFun SetupServiceFun) er
cls.callSet.Init()
if cls.IsNatsMode() {
cls.rpcNats.Init(cls.rpcMode.Nats.NatsUrl,cls.rpcMode.Nats.NoRandomize,cls.GetLocalNodeInfo().NodeId,cls.localNodeInfo.CompressBytesLen,cls,cluster.NotifyAllService)
cls.rpcNats.Init(cls.rpcMode.Nats.NatsUrl, cls.rpcMode.Nats.NoRandomize, cls.GetLocalNodeInfo().NodeId, cls.localNodeInfo.CompressBytesLen, cls, cluster.NotifyAllService)
cls.rpcServer = &cls.rpcNats
}else{
} else {
s := &rpc.Server{}
s.Init(cls.localNodeInfo.ListenAddr,cls.localNodeInfo.MaxRpcParamLen,cls.localNodeInfo.CompressBytesLen,cls)
s.Init(cls.localNodeInfo.ListenAddr, cls.localNodeInfo.MaxRpcParamLen, cls.localNodeInfo.CompressBytesLen, cls)
cls.rpcServer = s
}
//2.安装服务发现结点
err = cls.setupDiscovery(localNodeId, setupServiceFun)
if err != nil {
log.Error("setupDiscovery fail",log.ErrorAttr("err",err))
log.Error("setupDiscovery fail", log.ErrorAttr("err", err))
return err
}
service.RegRpcEventFun = cls.RegRpcEvent
@@ -274,16 +273,16 @@ func (cls *Cluster) FindRpcHandler(serviceName string) rpc.IRpcHandler {
return pService.GetRpcHandler()
}
func (cls *Cluster) getRpcClient(nodeId string) (*rpc.Client,bool) {
func (cls *Cluster) getRpcClient(nodeId string) (*rpc.Client, bool) {
c, ok := cls.mapRpc[nodeId]
if ok == false {
return nil,false
return nil, false
}
return c.client,c.nodeInfo.Retire
return c.client, c.nodeInfo.Retire
}
func (cls *Cluster) GetRpcClient(nodeId string) (*rpc.Client,bool) {
func (cls *Cluster) GetRpcClient(nodeId string) (*rpc.Client, bool) {
cls.locker.RLock()
defer cls.locker.RUnlock()
return cls.getRpcClient(nodeId)
@@ -293,9 +292,9 @@ func GetNodeIdByTemplateService(templateServiceName string, rpcClientList []*rpc
return GetCluster().GetNodeIdByTemplateService(templateServiceName, rpcClientList, filterRetire)
}
func GetRpcClient(nodeId string, serviceMethod string,filterRetire bool, clientList []*rpc.Client) (error, []*rpc.Client) {
func GetRpcClient(nodeId string, serviceMethod string, filterRetire bool, clientList []*rpc.Client) (error, []*rpc.Client) {
if nodeId != rpc.NodeIdNull {
pClient,retire := GetCluster().GetRpcClient(nodeId)
pClient, retire := GetCluster().GetRpcClient(nodeId)
if pClient == nil {
return fmt.Errorf("cannot find nodeid %s", nodeId), nil
}
@@ -305,7 +304,7 @@ func GetRpcClient(nodeId string, serviceMethod string,filterRetire bool, clientL
return fmt.Errorf("cannot find nodeid %s", nodeId), nil
}
clientList = append(clientList,pClient)
clientList = append(clientList, pClient)
return nil, clientList
}
@@ -315,7 +314,6 @@ func GetRpcClient(nodeId string, serviceMethod string,filterRetire bool, clientL
}
serviceName := serviceMethod[:findIndex]
//1.找到对应的rpcNodeid
return GetCluster().GetNodeIdByService(serviceName, clientList, filterRetire)
}
@@ -324,7 +322,7 @@ func GetRpcServer() rpc.IServer {
}
func (cls *Cluster) IsNodeConnected(nodeId string) bool {
pClient,_ := cls.GetRpcClient(nodeId)
pClient, _ := cls.GetRpcClient(nodeId)
return pClient != nil && pClient.IsConnected()
}
@@ -332,18 +330,18 @@ func (cls *Cluster) IsNodeRetire(nodeId string) bool {
cls.locker.RLock()
defer cls.locker.RUnlock()
_,retire :=cls.getRpcClient(nodeId)
_, retire := cls.getRpcClient(nodeId)
return retire
}
func (cls *Cluster) NotifyAllService(event event.IEvent){
func (cls *Cluster) NotifyAllService(event event.IEvent) {
cls.rpcEventLocker.Lock()
defer cls.rpcEventLocker.Unlock()
for serviceName, _ := range cls.mapServiceListenRpcEvent {
for serviceName := range cls.mapServiceListenRpcEvent {
ser := service.GetService(serviceName)
if ser == nil {
log.Error("cannot find service name "+serviceName)
log.Error("cannot find service name " + serviceName)
continue
}
@@ -403,7 +401,7 @@ func GetNodeByServiceName(serviceName string) map[string]struct{} {
}
mapNodeId := map[string]struct{}{}
for nodeId,_ := range mapNode {
for nodeId := range mapNode {
mapNodeId[nodeId] = struct{}{}
}
@@ -416,14 +414,14 @@ func GetNodeByTemplateServiceName(templateServiceName string) map[string]string
defer cluster.locker.RUnlock()
mapServiceName := cluster.mapTemplateServiceNode[templateServiceName]
mapNodeId := make(map[string]string,9)
mapNodeId := make(map[string]string, 9)
for serviceName := range mapServiceName {
mapNode, ok := cluster.mapServiceNode[serviceName]
if ok == false {
return nil
}
for nodeId,_ := range mapNode {
for nodeId := range mapNode {
mapNodeId[serviceName] = nodeId
}
}
@@ -435,55 +433,54 @@ func (cls *Cluster) GetGlobalCfg() interface{} {
return cls.globalCfg
}
func (cls *Cluster) ParseGlobalCfg(cfg interface{}) error{
if cls.globalCfg == nil {
func (cls *Cluster) ParseGlobalCfg(cfg interface{}) error {
if cls.globalCfg == nil {
return errors.New("no service configuration found")
}
rv := reflect.ValueOf(cls.globalCfg)
if rv.Kind() == reflect.Ptr && rv.IsNil() {
if rv.Kind() == reflect.Ptr && rv.IsNil() {
return errors.New("no service configuration found")
}
bytes,err := json.Marshal(cls.globalCfg)
bytes, err := json.Marshal(cls.globalCfg)
if err != nil {
return err
}
return json.Unmarshal(bytes,cfg)
return json.Unmarshal(bytes, cfg)
}
func (cls *Cluster) GetNodeInfo(nodeId string) (NodeInfo,bool) {
func (cls *Cluster) GetNodeInfo(nodeId string) (NodeInfo, bool) {
cls.locker.RLock()
defer cls.locker.RUnlock()
nodeInfo,ok:= cls.mapRpc[nodeId]
nodeInfo, ok := cls.mapRpc[nodeId]
if ok == false || nodeInfo == nil {
return NodeInfo{},false
return NodeInfo{}, false
}
return nodeInfo.nodeInfo,true
return nodeInfo.nodeInfo, true
}
func (cls *Cluster) CanDiscoveryService(fromMasterNodeId string,serviceName string) bool{
func (cls *Cluster) CanDiscoveryService(fromMasterNodeId string, serviceName string) bool {
canDiscovery := true
splitServiceName := strings.Split(serviceName,":")
splitServiceName := strings.Split(serviceName, ":")
if len(splitServiceName) == 2 {
serviceName = splitServiceName[0]
}
for i:=0;i<len(cls.GetLocalNodeInfo().DiscoveryService);i++{
for i := 0; i < len(cls.GetLocalNodeInfo().DiscoveryService); i++ {
masterNodeId := cls.GetLocalNodeInfo().DiscoveryService[i].MasterNodeId
//无效的配置，则跳过
if masterNodeId == rpc.NodeIdNull && len(cls.GetLocalNodeInfo().DiscoveryService[i].ServiceList)==0 {
if masterNodeId == rpc.NodeIdNull && len(cls.GetLocalNodeInfo().DiscoveryService[i].ServiceList) == 0 {
continue
}
canDiscovery = false
if masterNodeId == fromMasterNodeId || masterNodeId == rpc.NodeIdNull {
for _,discoveryService := range cls.GetLocalNodeInfo().DiscoveryService[i].ServiceList {
for _, discoveryService := range cls.GetLocalNodeInfo().DiscoveryService[i].ServiceList {
if discoveryService == serviceName {
return true
}
@@ -492,4 +489,4 @@ func (cls *Cluster) CanDiscoveryService(fromMasterNodeId string,serviceName stri
}
return canDiscovery
}
}

299
cluster/etcddiscovery.go
View File

@@ -1,6 +1,5 @@
package cluster
import (
"github.com/duanhf2012/origin/v2/event"
"github.com/duanhf2012/origin/v2/log"
@@ -23,33 +22,32 @@ import (
)
const originDir = "/origin"
const testKey = originDir+"/_inner/_test_7501f3ed-b716-44c2-0090-fc1ed0166d7a"
type etcdClientInfo struct {
watchKeys []string
leaseID clientv3.LeaseID
watchKeys []string
leaseID clientv3.LeaseID
keepAliveChan <-chan *clientv3.LeaseKeepAliveResponse
}
type EtcdDiscoveryService struct {
service.Service
funDelNode FunDelNode
funSetNode FunSetNode
localNodeId string
funDelNode FunDelNode
funSetNode FunSetNode
localNodeId string
byteLocalNodeInfo string
mapClient map[*clientv3.Client]*etcdClientInfo
isClose int32
bRetire bool
byteLocalNodeInfo string
mapClient map[*clientv3.Client]*etcdClientInfo
isClose int32
bRetire bool
mapDiscoveryNodeId map[string]map[string]struct{} //map[networkName]map[nodeId]
}
func getEtcdDiscovery() (IServiceDiscovery) {
func getEtcdDiscovery() IServiceDiscovery {
etcdDiscovery := &EtcdDiscoveryService{}
return etcdDiscovery
}
func (ed *EtcdDiscoveryService) InitDiscovery(localNodeId string,funDelNode FunDelNode,funSetNode FunSetNode) error {
func (ed *EtcdDiscoveryService) InitDiscovery(localNodeId string, funDelNode FunDelNode, funSetNode FunSetNode) error {
ed.localNodeId = localNodeId
ed.funDelNode = funDelNode
@@ -58,24 +56,24 @@ func (ed *EtcdDiscoveryService) InitDiscovery(localNodeId string,funDelNode FunD
return nil
}
const(
eeGets = 0
eePut = 1
const (
eeGets = 0
eePut = 1
eeDelete = 2
)
type etcdDiscoveryEvent struct {
typ int
typ int
watchKey string
Kvs []*mvccpb.KeyValue
Kvs []*mvccpb.KeyValue
}
func (ee *etcdDiscoveryEvent) GetEventType() event.EventType{
func (ee *etcdDiscoveryEvent) GetEventType() event.EventType {
return event.Sys_Event_EtcdDiscovery
}
func (ed *EtcdDiscoveryService) OnInit() error {
ed.mapClient = make(map[*clientv3.Client]*etcdClientInfo,1)
ed.mapClient = make(map[*clientv3.Client]*etcdClientInfo, 1)
ed.mapDiscoveryNodeId = make(map[string]map[string]struct{})
ed.GetEventProcessor().RegEventReceiverFunc(event.Sys_Event_EtcdDiscovery, ed.GetEventHandler(), ed.OnEtcdDiscovery)
@@ -90,28 +88,28 @@ func (ed *EtcdDiscoveryService) OnInit() error {
return errors.New("etcd discovery config is nil.")
}
for i:=0;i<len(etcdDiscoveryCfg.EtcdList);i++{
for i := 0; i < len(etcdDiscoveryCfg.EtcdList); i++ {
client, cerr := clientv3.New(clientv3.Config{
Endpoints: etcdDiscoveryCfg.EtcdList[i].Endpoints,
Endpoints: etcdDiscoveryCfg.EtcdList[i].Endpoints,
DialTimeout: etcdDiscoveryCfg.DialTimeoutMillisecond,
Logger: zap.NewNop(),
Logger: zap.NewNop(),
})
if cerr != nil {
log.Error("etcd discovery init fail",log.ErrorAttr("err",cerr))
log.Error("etcd discovery init fail", log.ErrorAttr("err", cerr))
return cerr
}
ctx,_:=context.WithTimeout(context.Background(),time.Second*3)
_,err = client.Leases(ctx)
ctx, _ := context.WithTimeout(context.Background(), time.Second*3)
_, err = client.Leases(ctx)
if err != nil {
log.Error("etcd discovery init fail",log.Any("endpoint",etcdDiscoveryCfg.EtcdList[i].Endpoints),log.ErrorAttr("err",err))
log.Error("etcd discovery init fail", log.Any("endpoint", etcdDiscoveryCfg.EtcdList[i].Endpoints), log.ErrorAttr("err", err))
return err
}
ec := &etcdClientInfo{}
for _, networkName := range etcdDiscoveryCfg.EtcdList[i].NetworkName {
ec.watchKeys = append(ec.watchKeys,fmt.Sprintf("%s/%s",originDir,networkName))
ec.watchKeys = append(ec.watchKeys, fmt.Sprintf("%s/%s", originDir, networkName))
}
ed.mapClient[client] = ec
@@ -120,36 +118,36 @@ func (ed *EtcdDiscoveryService) OnInit() error {
return nil
}
func (ed *EtcdDiscoveryService) getRegisterKey(watchkey string) string{
return watchkey+"/"+ed.localNodeId
func (ed *EtcdDiscoveryService) getRegisterKey(watchkey string) string {
return watchkey + "/" + ed.localNodeId
}
func (ed *EtcdDiscoveryService) registerServiceByClient(client *clientv3.Client,etcdClient *etcdClientInfo) {
func (ed *EtcdDiscoveryService) registerServiceByClient(client *clientv3.Client, etcdClient *etcdClientInfo) {
// 创建租约
var err error
var resp *clientv3.LeaseGrantResponse
resp, err = client.Grant(context.Background(), cluster.GetEtcdDiscovery().TTLSecond)
if err != nil {
log.Error("etcd registerService fail",log.ErrorAttr("err",err))
ed.tryRegisterService(client,etcdClient)
log.Error("etcd registerService fail", log.ErrorAttr("err", err))
ed.tryRegisterService(client, etcdClient)
return
}
etcdClient.leaseID = resp.ID
for _,watchKey:= range etcdClient.watchKeys {
for _, watchKey := range etcdClient.watchKeys {
// 注册服务节点到 etcd
_, err = client.Put(context.Background(), ed.getRegisterKey(watchKey), ed.byteLocalNodeInfo, clientv3.WithLease(resp.ID))
if err != nil {
log.Error("etcd Put fail",log.ErrorAttr("err",err))
ed.tryRegisterService(client,etcdClient)
log.Error("etcd Put fail", log.ErrorAttr("err", err))
ed.tryRegisterService(client, etcdClient)
return
}
}
etcdClient.keepAliveChan,err = client.KeepAlive(context.Background(), etcdClient.leaseID)
etcdClient.keepAliveChan, err = client.KeepAlive(context.Background(), etcdClient.leaseID)
if err != nil {
log.Error("etcd KeepAlive fail",log.ErrorAttr("err",err))
ed.tryRegisterService(client,etcdClient)
log.Error("etcd KeepAlive fail", log.ErrorAttr("err", err))
ed.tryRegisterService(client, etcdClient)
return
}
@@ -159,8 +157,8 @@ func (ed *EtcdDiscoveryService) registerServiceByClient(client *clientv3.Client,
case _, ok := <-etcdClient.keepAliveChan:
//log.Debug("ok",log.Any("addr",client.Endpoints()))
if !ok {
log.Error("etcd keepAliveChan fail",log.Any("watchKeys",etcdClient.watchKeys))
ed.tryRegisterService(client,etcdClient)
log.Error("etcd keepAliveChan fail", log.Any("watchKeys", etcdClient.watchKeys))
ed.tryRegisterService(client, etcdClient)
return
}
}
@@ -168,23 +166,22 @@ func (ed *EtcdDiscoveryService) registerServiceByClient(client *clientv3.Client,
}()
}
func (ed *EtcdDiscoveryService) tryRegisterService(client *clientv3.Client,etcdClient *etcdClientInfo){
func (ed *EtcdDiscoveryService) tryRegisterService(client *clientv3.Client, etcdClient *etcdClientInfo) {
if ed.isStop() {
return
}
ed.AfterFunc(time.Second*3, func(t *timer.Timer) {
ed.registerServiceByClient(client,etcdClient)
ed.registerServiceByClient(client, etcdClient)
})
}
func (ed *EtcdDiscoveryService) tryWatch(client *clientv3.Client,etcdClient *etcdClientInfo) {
func (ed *EtcdDiscoveryService) tryWatch(client *clientv3.Client, etcdClient *etcdClientInfo) {
if ed.isStop() {
return
}
ed.AfterFunc(time.Second*3, func(t *timer.Timer) {
ed.watchByClient(client,etcdClient)
ed.watchByClient(client, etcdClient)
})
}
@@ -196,9 +193,9 @@ func (ed *EtcdDiscoveryService) tryLaterRetire() {
})
}
func (ed *EtcdDiscoveryService) retire() error{
func (ed *EtcdDiscoveryService) retire() error {
//从etcd中更新
for c,ec := range ed.mapClient {
for c, ec := range ed.mapClient {
for _, watchKey := range ec.watchKeys {
// 注册服务节点到 etcd
_, err := c.Put(context.Background(), ed.getRegisterKey(watchKey), ed.byteLocalNodeInfo, clientv3.WithLease(ec.leaseID))
@@ -212,32 +209,32 @@ func (ed *EtcdDiscoveryService) retire() error{
return nil
}
func (ed *EtcdDiscoveryService) OnRetire(){
func (ed *EtcdDiscoveryService) OnRetire() {
ed.bRetire = true
ed.marshalNodeInfo()
if ed.retire()!= nil {
if ed.retire() != nil {
ed.tryLaterRetire()
}
}
func (ed *EtcdDiscoveryService) OnRelease(){
atomic.StoreInt32(&ed.isClose,1)
func (ed *EtcdDiscoveryService) OnRelease() {
atomic.StoreInt32(&ed.isClose, 1)
ed.close()
}
func (ed *EtcdDiscoveryService) isStop() bool{
func (ed *EtcdDiscoveryService) isStop() bool {
return atomic.LoadInt32(&ed.isClose) == 1
}
func (nd *EtcdDiscoveryService) OnStart() {
for c, ec := range nd.mapClient {
nd.tryRegisterService(c,ec)
nd.tryWatch(c,ec)
func (ed *EtcdDiscoveryService) OnStart() {
for c, ec := range ed.mapClient {
ed.tryRegisterService(c, ec)
ed.tryWatch(c, ec)
}
}
func (ed *EtcdDiscoveryService) marshalNodeInfo() error{
func (ed *EtcdDiscoveryService) marshalNodeInfo() error {
nInfo := cluster.GetLocalNodeInfo()
var nodeInfo rpc.NodeInfo
nodeInfo.NodeId = nInfo.NodeId
@@ -246,15 +243,15 @@ func (ed *EtcdDiscoveryService) marshalNodeInfo() error{
nodeInfo.PublicServiceList = nInfo.PublicServiceList
nodeInfo.MaxRpcParamLen = nInfo.MaxRpcParamLen
byteLocalNodeInfo,err := proto.Marshal(&nodeInfo)
if err ==nil{
byteLocalNodeInfo, err := proto.Marshal(&nodeInfo)
if err == nil {
ed.byteLocalNodeInfo = string(byteLocalNodeInfo)
}
return err
}
func (ed *EtcdDiscoveryService) setNodeInfo(networkName string,nodeInfo *rpc.NodeInfo) bool{
func (ed *EtcdDiscoveryService) setNodeInfo(networkName string, nodeInfo *rpc.NodeInfo) bool {
if nodeInfo == nil || nodeInfo.Private == true || nodeInfo.NodeId == ed.localNodeId {
return false
}
@@ -262,8 +259,8 @@ func (ed *EtcdDiscoveryService) setNodeInfo(networkName string,nodeInfo *rpc.Nod
//筛选关注的服务
var discoverServiceSlice = make([]string, 0, 24)
for _, pubService := range nodeInfo.PublicServiceList {
if cluster.CanDiscoveryService(networkName,pubService) == true {
discoverServiceSlice = append(discoverServiceSlice,pubService)
if cluster.CanDiscoveryService(networkName, pubService) == true {
discoverServiceSlice = append(discoverServiceSlice, pubService)
}
}
@@ -285,20 +282,20 @@ func (ed *EtcdDiscoveryService) setNodeInfo(networkName string,nodeInfo *rpc.Nod
return true
}
func (ed *EtcdDiscoveryService) close(){
func (ed *EtcdDiscoveryService) close() {
for c, ec := range ed.mapClient {
if _, err := c.Revoke(context.Background(), ec.leaseID); err != nil {
log.Error("etcd Revoke fail",log.ErrorAttr("err",err))
log.Error("etcd Revoke fail", log.ErrorAttr("err", err))
}
c.Watcher.Close()
err := c.Close()
if err != nil {
log.Error("etcd Close fail",log.ErrorAttr("err",err))
log.Error("etcd Close fail", log.ErrorAttr("err", err))
}
}
}
func (ed *EtcdDiscoveryService) getServices(client *clientv3.Client,etcdClient *etcdClientInfo,watchKey string) bool {
func (ed *EtcdDiscoveryService) getServices(client *clientv3.Client, etcdClient *etcdClientInfo, watchKey string) bool {
// 根据前缀获取现有的key
resp, err := client.Get(context.Background(), watchKey, clientv3.WithPrefix())
if err != nil {
@@ -308,35 +305,35 @@ func (ed *EtcdDiscoveryService) getServices(client *clientv3.Client,etcdClient *
}
// 遍历获取得到的k和v
ed.notifyGets(watchKey,resp.Kvs)
ed.notifyGets(watchKey, resp.Kvs)
return true
}
func (ed *EtcdDiscoveryService) watchByClient(client *clientv3.Client,etcdClient *etcdClientInfo){
func (ed *EtcdDiscoveryService) watchByClient(client *clientv3.Client, etcdClient *etcdClientInfo) {
//先关闭所有的watcher
for _, watchKey := range etcdClient.watchKeys {
// 监视前缀，修改变更server
go ed.watcher(client,etcdClient, watchKey)
go ed.watcher(client, etcdClient, watchKey)
}
}
// watcher 监听Key的前缀
func (ed *EtcdDiscoveryService) watcher(client *clientv3.Client,etcdClient *etcdClientInfo,watchKey string) {
func (ed *EtcdDiscoveryService) watcher(client *clientv3.Client, etcdClient *etcdClientInfo, watchKey string) {
defer func() {
if r := recover(); r != nil {
buf := make([]byte, 4096)
l := runtime.Stack(buf, false)
errString := fmt.Sprint(r)
log.Dump(string(buf[:l]),log.String("error",errString))
log.Dump(string(buf[:l]), log.String("error", errString))
ed.tryWatch(client,etcdClient)
ed.tryWatch(client, etcdClient)
}
}()
rch := client.Watch(context.Background(), watchKey, clientv3.WithPrefix())
if ed.getServices(client,etcdClient,watchKey) == false {
if ed.getServices(client, etcdClient, watchKey) == false {
return
}
@@ -344,68 +341,68 @@ func (ed *EtcdDiscoveryService) watcher(client *clientv3.Client,etcdClient *etcd
for _, ev := range wresp.Events {
switch ev.Type {
case clientv3.EventTypePut: // 修改或者新增
ed.notifyPut(watchKey,ev.Kv)
ed.notifyPut(watchKey, ev.Kv)
case clientv3.EventTypeDelete: // 删除
ed.notifyDelete(watchKey,ev.Kv)
ed.notifyDelete(watchKey, ev.Kv)
}
}
}
ed.tryWatch(client,etcdClient)
ed.tryWatch(client, etcdClient)
}
func (ed *EtcdDiscoveryService) setNode(netWorkName string,byteNode []byte) string{
func (ed *EtcdDiscoveryService) setNode(netWorkName string, byteNode []byte) string {
var nodeInfo rpc.NodeInfo
err := proto.Unmarshal(byteNode,&nodeInfo)
err := proto.Unmarshal(byteNode, &nodeInfo)
if err != nil {
log.Error("Unmarshal fail",log.String("netWorkName",netWorkName),log.ErrorAttr("err",err))
log.Error("Unmarshal fail", log.String("netWorkName", netWorkName), log.ErrorAttr("err", err))
return ""
}
ed.setNodeInfo(netWorkName,&nodeInfo)
ed.setNodeInfo(netWorkName, &nodeInfo)
return nodeInfo.NodeId
}
func (ed *EtcdDiscoveryService) delNode(fullKey string) string{
func (ed *EtcdDiscoveryService) delNode(fullKey string) string {
nodeId := ed.getNodeId(fullKey)
if nodeId == ed.localNodeId {
return ""
}
ed.funDelNode(nodeId)
ed.funDelNode(nodeId)
return nodeId
}
func (ed *EtcdDiscoveryService) getNetworkNameByWatchKey(watchKey string) string{
return watchKey[strings.LastIndex(watchKey,"/")+1:]
func (ed *EtcdDiscoveryService) getNetworkNameByWatchKey(watchKey string) string {
return watchKey[strings.LastIndex(watchKey, "/")+1:]
}
func (ed *EtcdDiscoveryService) getNetworkNameByFullKey(fullKey string) string{
return fullKey[len(originDir)+1:strings.LastIndex(fullKey,"/")]
func (ed *EtcdDiscoveryService) getNetworkNameByFullKey(fullKey string) string {
return fullKey[len(originDir)+1 : strings.LastIndex(fullKey, "/")]
}
func (ed *EtcdDiscoveryService) getNodeId(fullKey string) string{
return fullKey[strings.LastIndex(fullKey,"/")+1:]
func (ed *EtcdDiscoveryService) getNodeId(fullKey string) string {
return fullKey[strings.LastIndex(fullKey, "/")+1:]
}
func (ed *EtcdDiscoveryService) OnEtcdDiscovery(ev event.IEvent){
func (ed *EtcdDiscoveryService) OnEtcdDiscovery(ev event.IEvent) {
disEvent := ev.(*etcdDiscoveryEvent)
switch disEvent.typ {
case eeGets:
ed.OnEventGets(disEvent.watchKey,disEvent.Kvs)
ed.OnEventGets(disEvent.watchKey, disEvent.Kvs)
case eePut:
if len(disEvent.Kvs) == 1 {
ed.OnEventPut(disEvent.watchKey,disEvent.Kvs[0])
ed.OnEventPut(disEvent.watchKey, disEvent.Kvs[0])
}
case eeDelete:
if len(disEvent.Kvs) == 1 {
ed.OnEventDelete(disEvent.watchKey,disEvent.Kvs[0])
ed.OnEventDelete(disEvent.watchKey, disEvent.Kvs[0])
}
}
}
func (ed *EtcdDiscoveryService) notifyGets(watchKey string,Kvs []*mvccpb.KeyValue) {
func (ed *EtcdDiscoveryService) notifyGets(watchKey string, Kvs []*mvccpb.KeyValue) {
var ev etcdDiscoveryEvent
ev.typ = eeGets
ev.watchKey = watchKey
@@ -413,54 +410,54 @@ func (ed *EtcdDiscoveryService) notifyGets(watchKey string,Kvs []*mvccpb.KeyValu
ed.NotifyEvent(&ev)
}
func (ed *EtcdDiscoveryService) notifyPut(watchKey string,Kvs *mvccpb.KeyValue) {
func (ed *EtcdDiscoveryService) notifyPut(watchKey string, Kvs *mvccpb.KeyValue) {
var ev etcdDiscoveryEvent
ev.typ = eePut
ev.watchKey = watchKey
ev.Kvs = append(ev.Kvs,Kvs)
ev.Kvs = append(ev.Kvs, Kvs)
ed.NotifyEvent(&ev)
}
func (ed *EtcdDiscoveryService) notifyDelete(watchKey string,Kvs *mvccpb.KeyValue) {
func (ed *EtcdDiscoveryService) notifyDelete(watchKey string, Kvs *mvccpb.KeyValue) {
var ev etcdDiscoveryEvent
ev.typ = eeDelete
ev.watchKey = watchKey
ev.Kvs = append(ev.Kvs,Kvs)
ev.Kvs = append(ev.Kvs, Kvs)
ed.NotifyEvent(&ev)
}
func (ed *EtcdDiscoveryService) OnEventGets(watchKey string,Kvs []*mvccpb.KeyValue) {
mapNode := make(map[string]struct{},32)
func (ed *EtcdDiscoveryService) OnEventGets(watchKey string, Kvs []*mvccpb.KeyValue) {
mapNode := make(map[string]struct{}, 32)
for _, kv := range Kvs {
nodeId := ed.setNode(ed.getNetworkNameByFullKey(string(kv.Key)), kv.Value)
mapNode[nodeId] = struct{}{}
ed.addNodeId(watchKey,nodeId)
ed.addNodeId(watchKey, nodeId)
}
// 此段代码为遍历并删除过期节点的逻辑。
// 对于mapDiscoveryNodeId中与watchKey关联的所有节点ID，遍历该集合。
// 如果某个节点ID不在mapNode中且不是本地节点ID，则调用funDelNode函数删除该节点。
mapLastNodeId := ed.mapDiscoveryNodeId[watchKey] // 根据watchKey获取对应的节点ID集合
for nodeId := range mapLastNodeId { // 遍历所有节点ID
if _,ok := mapNode[nodeId];ok == false && nodeId != ed.localNodeId { // 检查节点是否不存在于mapNode且不是本地节点
ed.funDelNode(nodeId) // 调用函数删除该节点
delete(ed.mapDiscoveryNodeId[watchKey],nodeId)
}
for nodeId := range mapLastNodeId { // 遍历所有节点ID
if _, ok := mapNode[nodeId]; ok == false && nodeId != ed.localNodeId { // 检查节点是否不存在于mapNode且不是本地节点
ed.funDelNode(nodeId) // 调用函数删除该节点
delete(ed.mapDiscoveryNodeId[watchKey], nodeId)
}
}
}
func (ed *EtcdDiscoveryService) OnEventPut(watchKey string,Kv *mvccpb.KeyValue) {
func (ed *EtcdDiscoveryService) OnEventPut(watchKey string, Kv *mvccpb.KeyValue) {
nodeId := ed.setNode(ed.getNetworkNameByFullKey(string(Kv.Key)), Kv.Value)
ed.addNodeId(watchKey,nodeId)
ed.addNodeId(watchKey, nodeId)
}
func (ed *EtcdDiscoveryService) OnEventDelete(watchKey string,Kv *mvccpb.KeyValue) {
func (ed *EtcdDiscoveryService) OnEventDelete(watchKey string, Kv *mvccpb.KeyValue) {
nodeId := ed.delNode(string(Kv.Key))
delete(ed.mapDiscoveryNodeId[watchKey],nodeId)
delete(ed.mapDiscoveryNodeId[watchKey], nodeId)
}
func (ed *EtcdDiscoveryService) addNodeId(watchKey string,nodeId string) {
if _,ok := ed.mapDiscoveryNodeId[watchKey];ok == false {
func (ed *EtcdDiscoveryService) addNodeId(watchKey string, nodeId string) {
if _, ok := ed.mapDiscoveryNodeId[watchKey]; ok == false {
ed.mapDiscoveryNodeId[watchKey] = make(map[string]struct{})
}
@@ -472,50 +469,50 @@ func (ed *EtcdDiscoveryService) OnNodeDisconnect(nodeId string) {
cluster.DiscardNode(nodeId)
}
func (ed *EtcdDiscoveryService) RPC_ServiceRecord(etcdServiceRecord *service.EtcdServiceRecordEvent,empty *service.Empty) error{
var client *clientv3.Client
func (ed *EtcdDiscoveryService) RPC_ServiceRecord(etcdServiceRecord *service.EtcdServiceRecordEvent, empty *service.Empty) error {
var client *clientv3.Client
//写入到etcd中
for c, info := range ed.mapClient{
for _,watchKey := range info.watchKeys {
if ed.getNetworkNameByWatchKey(watchKey) == etcdServiceRecord.NetworkName {
client = c
break
}
//写入到etcd中
for c, info := range ed.mapClient {
for _, watchKey := range info.watchKeys {
if ed.getNetworkNameByWatchKey(watchKey) == etcdServiceRecord.NetworkName {
client = c
break
}
}
}
if client == nil {
log.Error("etcd record fail,cannot find network name",log.String("networkName",etcdServiceRecord.NetworkName))
return errors.New("annot find network name")
}
if client == nil {
log.Error("etcd record fail,cannot find network name", log.String("networkName", etcdServiceRecord.NetworkName))
return errors.New("annot find network name")
}
var lg *clientv3.LeaseGrantResponse
var err error
var lg *clientv3.LeaseGrantResponse
var err error
if etcdServiceRecord.TTLSecond > 0 {
ctx,_:=context.WithTimeout(context.Background(),time.Second*3)
lg, err = client.Grant(ctx, etcdServiceRecord.TTLSecond)
if err != nil {
log.Error("etcd record fail,cannot grant lease",log.ErrorAttr("err",err))
return errors.New("cannot grant lease")
}
}
if lg != nil {
ctx,_:=context.WithTimeout(context.Background(),time.Second*3)
_, err = client.Put(ctx, path.Join(originDir,etcdServiceRecord.RecordKey),etcdServiceRecord.RecordInfo, clientv3.WithLease(lg.ID))
if err != nil {
log.Error("etcd record fail,cannot put record",log.ErrorAttr("err",err))
}
return errors.New("cannot put record")
}
_,err = client.Put(context.Background(), path.Join(originDir,etcdServiceRecord.RecordKey),etcdServiceRecord.RecordInfo)
if etcdServiceRecord.TTLSecond > 0 {
ctx, _ := context.WithTimeout(context.Background(), time.Second*3)
lg, err = client.Grant(ctx, etcdServiceRecord.TTLSecond)
if err != nil {
log.Error("etcd record fail,cannot put record",log.ErrorAttr("err",err))
return errors.New("cannot put record")
log.Error("etcd record fail,cannot grant lease", log.ErrorAttr("err", err))
return errors.New("cannot grant lease")
}
}
return nil
if lg != nil {
ctx, _ := context.WithTimeout(context.Background(), time.Second*3)
_, err = client.Put(ctx, path.Join(originDir, etcdServiceRecord.RecordKey), etcdServiceRecord.RecordInfo, clientv3.WithLease(lg.ID))
if err != nil {
log.Error("etcd record fail,cannot put record", log.ErrorAttr("err", err))
}
return errors.New("cannot put record")
}
_, err = client.Put(context.Background(), path.Join(originDir, etcdServiceRecord.RecordKey), etcdServiceRecord.RecordInfo)
if err != nil {
log.Error("etcd record fail,cannot put record", log.ErrorAttr("err", err))
return errors.New("cannot put record")
}
return nil
}

36
cluster/nodettl.go
View File

@@ -1,53 +1,53 @@
package cluster
import (
"time"
"container/list"
"time"
)
type nodeTTL struct {
nodeId string
nodeId string
refreshTime time.Time
}
type nodeSetTTL struct {
l *list.List
l *list.List
mapElement map[string]*list.Element
ttl time.Duration
ttl time.Duration
}
func (ns *nodeSetTTL) init(ttl time.Duration) {
ns.ttl = ttl
ns.mapElement = make(map[string]*list.Element,32)
ns.mapElement = make(map[string]*list.Element, 32)
ns.l = list.New()
}
func (ns *nodeSetTTL) removeNode(nodeId string) {
ele,ok:=ns.mapElement[nodeId]
ele, ok := ns.mapElement[nodeId]
if ok == false {
return
}
ns.l.Remove(ele)
delete(ns.mapElement,nodeId)
delete(ns.mapElement, nodeId)
}
func (ns *nodeSetTTL) addAndRefreshNode(nodeId string){
ele,ok:=ns.mapElement[nodeId]
func (ns *nodeSetTTL) addAndRefreshNode(nodeId string) {
ele, ok := ns.mapElement[nodeId]
if ok == false {
ele = ns.l.PushBack(nodeId)
ele.Value = &nodeTTL{nodeId,time.Now()}
ele.Value = &nodeTTL{nodeId, time.Now()}
ns.mapElement[nodeId] = ele
return
}
ele.Value.(*nodeTTL).refreshTime = time.Now()
ele.Value.(*nodeTTL).refreshTime = time.Now()
ns.l.MoveToBack(ele)
}
func (ns *nodeSetTTL) checkTTL(cb func(nodeIdList []string)){
nodeIdList := []string{}
for{
func (ns *nodeSetTTL) checkTTL(cb func(nodeIdList []string)) {
var nodeIdList []string
for {
f := ns.l.Front()
if f == nil {
break
@@ -55,17 +55,17 @@ func (ns *nodeSetTTL) checkTTL(cb func(nodeIdList []string)){
nt := f.Value.(*nodeTTL)
if time.Now().Sub(nt.refreshTime) > ns.ttl {
nodeIdList = append(nodeIdList,nt.nodeId)
}else{
nodeIdList = append(nodeIdList, nt.nodeId)
} else {
break
}
//删除结点
ns.l.Remove(f)
delete(ns.mapElement,nt.nodeId)
delete(ns.mapElement, nt.nodeId)
}
if len(nodeIdList) >0 {
if len(nodeIdList) > 0 {
cb(nodeIdList)
}
}

150
cluster/origindiscovery.go
View File

@@ -14,10 +14,9 @@ const OriginDiscoveryMasterName = "DiscoveryMaster"
const OriginDiscoveryClientName = "DiscoveryClient"
const RegServiceDiscover = OriginDiscoveryMasterName + ".RPC_RegServiceDiscover"
const SubServiceDiscover = OriginDiscoveryClientName + ".RPC_SubServiceDiscover"
const AddSubServiceDiscover = OriginDiscoveryMasterName + ".RPC_AddSubServiceDiscover"
const NodeRetireRpcMethod = OriginDiscoveryMasterName+".RPC_NodeRetire"
const RpcPingMethod = OriginDiscoveryMasterName+".RPC_Ping"
const UnRegServiceDiscover = OriginDiscoveryMasterName+".RPC_UnRegServiceDiscover"
const NodeRetireRpcMethod = OriginDiscoveryMasterName + ".RPC_NodeRetire"
const RpcPingMethod = OriginDiscoveryMasterName + ".RPC_Ping"
const UnRegServiceDiscover = OriginDiscoveryMasterName + ".RPC_UnRegServiceDiscover"
type OriginDiscoveryMaster struct {
service.Service
@@ -31,12 +30,12 @@ type OriginDiscoveryMaster struct {
type OriginDiscoveryClient struct {
service.Service
funDelNode FunDelNode
funSetNode FunSetNode
localNodeId string
funDelNode FunDelNode
funSetNode FunSetNode
localNodeId string
mapDiscovery map[string]map[string][]string //map[masterNodeId]map[nodeId]struct{}
bRetire bool
bRetire bool
isRegisterOk bool
}
@@ -58,14 +57,14 @@ func (ds *OriginDiscoveryMaster) isRegNode(nodeId string) bool {
}
func (ds *OriginDiscoveryMaster) updateNodeInfo(nInfo *rpc.NodeInfo) {
if _,ok:= ds.mapNodeInfo[nInfo.NodeId];ok == false {
if _, ok := ds.mapNodeInfo[nInfo.NodeId]; ok == false {
return
}
nodeInfo := proto.Clone(nInfo).(*rpc.NodeInfo)
for i:=0;i<len(ds.nodeInfo);i++ {
for i := 0; i < len(ds.nodeInfo); i++ {
if ds.nodeInfo[i].NodeId == nodeInfo.NodeId {
ds.nodeInfo[i] = nodeInfo
ds.nodeInfo[i] = nodeInfo
break
}
}
@@ -87,19 +86,19 @@ func (ds *OriginDiscoveryMaster) addNodeInfo(nInfo *rpc.NodeInfo) {
}
func (ds *OriginDiscoveryMaster) removeNodeInfo(nodeId string) {
if _,ok:= ds.mapNodeInfo[nodeId];ok == false {
if _, ok := ds.mapNodeInfo[nodeId]; ok == false {
return
}
for i:=0;i<len(ds.nodeInfo);i++ {
for i := 0; i < len(ds.nodeInfo); i++ {
if ds.nodeInfo[i].NodeId == nodeId {
ds.nodeInfo = append(ds.nodeInfo[:i],ds.nodeInfo[i+1:]...)
ds.nodeInfo = append(ds.nodeInfo[:i], ds.nodeInfo[i+1:]...)
break
}
}
ds.nsTTL.removeNode(nodeId)
delete(ds.mapNodeInfo,nodeId)
delete(ds.mapNodeInfo, nodeId)
}
func (ds *OriginDiscoveryMaster) OnInit() error {
@@ -107,26 +106,26 @@ func (ds *OriginDiscoveryMaster) OnInit() error {
ds.RegNodeConnListener(ds)
ds.RegNatsConnListener(ds)
ds.nsTTL.init(time.Duration(cluster.GetOriginDiscovery().TTLSecond)*time.Second)
ds.nsTTL.init(time.Duration(cluster.GetOriginDiscovery().TTLSecond) * time.Second)
return nil
}
func (ds *OriginDiscoveryMaster) checkTTL(){
func (ds *OriginDiscoveryMaster) checkTTL() {
if cluster.IsNatsMode() == false {
return
}
interval := time.Duration(cluster.GetOriginDiscovery().TTLSecond)*time.Second
interval = interval /3 /2
interval := time.Duration(cluster.GetOriginDiscovery().TTLSecond) * time.Second
interval = interval / 3 / 2
if interval < time.Second {
interval = time.Second
}
ds.NewTicker(interval,func(t *timer.Ticker){
ds.NewTicker(interval, func(t *timer.Ticker) {
ds.nsTTL.checkTTL(func(nodeIdList []string) {
for _,nodeId := range nodeIdList {
log.Info("TTL expiry",log.String("nodeId",nodeId))
for _, nodeId := range nodeIdList {
log.Info("TTL expiry", log.String("nodeId", nodeId))
ds.OnNodeDisconnect(nodeId)
}
})
@@ -147,16 +146,16 @@ func (ds *OriginDiscoveryMaster) OnStart() {
ds.checkTTL()
}
func (dc *OriginDiscoveryMaster) OnNatsConnected(){
func (ds *OriginDiscoveryMaster) OnNatsConnected() {
//向所有的节点同步服务发现信息
var notifyDiscover rpc.SubscribeDiscoverNotify
notifyDiscover.IsFull = true
notifyDiscover.NodeInfo = dc.nodeInfo
notifyDiscover.NodeInfo = ds.nodeInfo
notifyDiscover.MasterNodeId = cluster.GetLocalNodeInfo().NodeId
dc.RpcCastGo(SubServiceDiscover, &notifyDiscover)
ds.RpcCastGo(SubServiceDiscover, &notifyDiscover)
}
func (dc *OriginDiscoveryMaster) OnNatsDisconnect(){
func (ds *OriginDiscoveryMaster) OnNatsDisconnect() {
}
func (ds *OriginDiscoveryMaster) OnNodeConnected(nodeId string) {
@@ -188,7 +187,7 @@ func (ds *OriginDiscoveryMaster) OnNodeDisconnect(nodeId string) {
}
func (ds *OriginDiscoveryMaster) RpcCastGo(serviceMethod string, args interface{}) {
for nodeId, _ := range ds.mapNodeInfo {
for nodeId := range ds.mapNodeInfo {
if nodeId == cluster.GetLocalNodeInfo().NodeId {
continue
}
@@ -198,7 +197,7 @@ func (ds *OriginDiscoveryMaster) RpcCastGo(serviceMethod string, args interface{
}
func (ds *OriginDiscoveryMaster) RPC_Ping(req *rpc.Ping, res *rpc.Pong) error {
if ds.isRegNode(req.NodeId) == false{
if ds.isRegNode(req.NodeId) == false {
res.Ok = false
return nil
}
@@ -208,8 +207,8 @@ func (ds *OriginDiscoveryMaster) RPC_Ping(req *rpc.Ping, res *rpc.Pong) error {
return nil
}
func (ds *OriginDiscoveryMaster) RPC_NodeRetire(req *rpc.NodeRetireReq, res *rpc.Empty) error {
log.Info("node is retire",log.String("nodeId",req.NodeInfo.NodeId),log.Bool("retire",req.NodeInfo.Retire))
func (ds *OriginDiscoveryMaster) RPC_NodeRetire(req *rpc.NodeRetireReq, _ *rpc.Empty) error {
log.Info("node is retire", log.String("nodeId", req.NodeInfo.NodeId), log.Bool("retire", req.NodeInfo.Retire))
ds.updateNodeInfo(req.NodeInfo)
@@ -259,15 +258,14 @@ func (ds *OriginDiscoveryMaster) RPC_RegServiceDiscover(req *rpc.RegServiceDisco
//加入到本地Cluster模块中，将连接该结点
cluster.serviceDiscoverySetNodeInfo(&nodeInfo)
res.IsFull = true
res.NodeInfo = ds.nodeInfo
res.MasterNodeId = cluster.GetLocalNodeInfo().NodeId
return nil
}
func (ds *OriginDiscoveryMaster) RPC_UnRegServiceDiscover(req *rpc.UnRegServiceDiscoverReq, res *rpc.Empty) error {
log.Debug("RPC_UnRegServiceDiscover",log.String("nodeId",req.NodeId))
func (ds *OriginDiscoveryMaster) RPC_UnRegServiceDiscover(req *rpc.UnRegServiceDiscoverReq, _ *rpc.Empty) error {
log.Debug("RPC_UnRegServiceDiscover", log.String("nodeId", req.NodeId))
ds.OnNodeDisconnect(req.NodeId)
return nil
}
@@ -282,7 +280,7 @@ func (dc *OriginDiscoveryClient) OnInit() error {
return nil
}
func (dc *OriginDiscoveryClient) addMasterNode(masterNodeId string, nodeId string,serviceList []string) {
func (dc *OriginDiscoveryClient) addMasterNode(masterNodeId string, nodeId string, serviceList []string) {
_, ok := dc.mapDiscovery[masterNodeId]
if ok == false {
dc.mapDiscovery[masterNodeId] = map[string][]string{}
@@ -290,7 +288,7 @@ func (dc *OriginDiscoveryClient) addMasterNode(masterNodeId string, nodeId strin
dc.mapDiscovery[masterNodeId][nodeId] = serviceList
}
func (dc *OriginDiscoveryClient) getNodePublicService(masterNodeId string,nodeId string) []string{
func (dc *OriginDiscoveryClient) getNodePublicService(masterNodeId string, nodeId string) []string {
mapNodeId, ok := dc.mapDiscovery[masterNodeId]
if ok == false {
return nil
@@ -320,28 +318,28 @@ func (dc *OriginDiscoveryClient) findNodeId(nodeId string) bool {
return false
}
func (dc *OriginDiscoveryClient) ping(){
interval := time.Duration(cluster.GetOriginDiscovery().TTLSecond)*time.Second
interval = interval /3
func (dc *OriginDiscoveryClient) ping() {
interval := time.Duration(cluster.GetOriginDiscovery().TTLSecond) * time.Second
interval = interval / 3
if interval < time.Second {
interval = time.Second
}
dc.NewTicker(interval,func(t *timer.Ticker){
if cluster.IsNatsMode() == false || dc.isRegisterOk == false{
dc.NewTicker(interval, func(t *timer.Ticker) {
if cluster.IsNatsMode() == false || dc.isRegisterOk == false {
return
}
var ping rpc.Ping
ping.NodeId = cluster.GetLocalNodeInfo().NodeId
masterNodes := GetCluster().GetOriginDiscovery().MasterNodeList
for i:=0;i<len(masterNodes);i++ {
for i := 0; i < len(masterNodes); i++ {
if masterNodes[i].NodeId == cluster.GetLocalNodeInfo().NodeId {
continue
}
masterNodeId := masterNodes[i].NodeId
dc.AsyncCallNodeWithTimeout(3*time.Second,masterNodeId,RpcPingMethod,&ping, func(empty *rpc.Pong,err error) {
if err == nil && empty.Ok == false{
dc.AsyncCallNodeWithTimeout(3*time.Second, masterNodeId, RpcPingMethod, &ping, func(empty *rpc.Pong, err error) {
if err == nil && empty.Ok == false {
//断开master重
dc.regServiceDiscover(masterNodeId)
}
@@ -350,7 +348,6 @@ func (dc *OriginDiscoveryClient) ping(){
})
}
func (dc *OriginDiscoveryClient) OnStart() {
//2.添加并连接发现主结点
dc.addDiscoveryMaster()
@@ -381,8 +378,8 @@ func (dc *OriginDiscoveryClient) fullCompareDiffNode(masterNodeId string, mapNod
}
//本地任何Master都不存在的，放到diffNodeIdSlice
for nodeId, _ := range mapNodeId {
_, ok := mapNodeInfo[nodeId]
for nodeId := range mapNodeId {
_, ok = mapNodeInfo[nodeId]
if ok == false {
diffNodeIdSlice = append(diffNodeIdSlice, nodeId)
}
@@ -391,7 +388,7 @@ func (dc *OriginDiscoveryClient) fullCompareDiffNode(masterNodeId string, mapNod
return diffNodeIdSlice
}
//订阅发现的服务通知
// RPC_SubServiceDiscover 订阅发现的服务通知
func (dc *OriginDiscoveryClient) RPC_SubServiceDiscover(req *rpc.SubscribeDiscoverNotify) error {
mapNodeInfo := map[string]*rpc.NodeInfo{}
for _, nodeInfo := range req.NodeInfo {
@@ -445,8 +442,8 @@ func (dc *OriginDiscoveryClient) RPC_SubServiceDiscover(req *rpc.SubscribeDiscov
//设置新结点
for _, nodeInfo := range mapNodeInfo {
dc.addMasterNode(req.MasterNodeId, nodeInfo.NodeId,nodeInfo.PublicServiceList)
bSet := dc.setNodeInfo(req.MasterNodeId,nodeInfo)
dc.addMasterNode(req.MasterNodeId, nodeInfo.NodeId, nodeInfo.PublicServiceList)
bSet := dc.setNodeInfo(req.MasterNodeId, nodeInfo)
if bSet == false {
continue
}
@@ -459,7 +456,7 @@ func (dc *OriginDiscoveryClient) OnNodeConnected(nodeId string) {
dc.regServiceDiscover(nodeId)
}
func (dc *OriginDiscoveryClient) OnRelease(){
func (dc *OriginDiscoveryClient) OnRelease() {
log.Debug("OriginDiscoveryClient")
//取消注册
@@ -467,47 +464,47 @@ func (dc *OriginDiscoveryClient) OnRelease(){
nodeRetireReq.NodeId = cluster.GetLocalNodeInfo().NodeId
masterNodeList := cluster.GetOriginDiscovery()
for i:=0;i<len(masterNodeList.MasterNodeList);i++{
for i := 0; i < len(masterNodeList.MasterNodeList); i++ {
if masterNodeList.MasterNodeList[i].NodeId == cluster.GetLocalNodeInfo().NodeId {
continue
}
err := dc.CallNodeWithTimeout(3*time.Second,masterNodeList.MasterNodeList[i].NodeId,UnRegServiceDiscover,&nodeRetireReq,&rpc.Empty{})
if err!= nil {
log.Error("call "+UnRegServiceDiscover+" is fail",log.ErrorAttr("err",err))
err := dc.CallNodeWithTimeout(3*time.Second, masterNodeList.MasterNodeList[i].NodeId, UnRegServiceDiscover, &nodeRetireReq, &rpc.Empty{})
if err != nil {
log.Error("call "+UnRegServiceDiscover+" is fail", log.ErrorAttr("err", err))
}
}
}
func (dc *OriginDiscoveryClient) OnRetire(){
func (dc *OriginDiscoveryClient) OnRetire() {
dc.bRetire = true
masterNodeList := cluster.GetOriginDiscovery()
for i:=0;i<len(masterNodeList.MasterNodeList);i++{
for i := 0; i < len(masterNodeList.MasterNodeList); i++ {
var nodeRetireReq rpc.NodeRetireReq
nodeRetireReq.NodeInfo = &rpc.NodeInfo{}
nodeRetireReq.NodeInfo.NodeId = cluster.localNodeInfo.NodeId
nodeRetireReq.NodeInfo.ListenAddr = cluster.localNodeInfo.ListenAddr
nodeRetireReq.NodeInfo.MaxRpcParamLen = cluster.localNodeInfo.MaxRpcParamLen
nodeRetireReq.NodeInfo.PublicServiceList = cluster.localNodeInfo.PublicServiceList
nodeRetireReq.NodeInfo.PublicServiceList = cluster.localNodeInfo.PublicServiceList
nodeRetireReq.NodeInfo.Retire = dc.bRetire
nodeRetireReq.NodeInfo.Private = cluster.localNodeInfo.Private
err := dc.GoNode(masterNodeList.MasterNodeList[i].NodeId,NodeRetireRpcMethod,&nodeRetireReq)
if err!= nil {
log.Error("call "+NodeRetireRpcMethod+" is fail",log.ErrorAttr("err",err))
err := dc.GoNode(masterNodeList.MasterNodeList[i].NodeId, NodeRetireRpcMethod, &nodeRetireReq)
if err != nil {
log.Error("call "+NodeRetireRpcMethod+" is fail", log.ErrorAttr("err", err))
}
}
}
func (dc *OriginDiscoveryClient) tryRegServiceDiscover(nodeId string){
func (dc *OriginDiscoveryClient) tryRegServiceDiscover(nodeId string) {
dc.AfterFunc(time.Second*3, func(timer *timer.Timer) {
dc.regServiceDiscover(nodeId)
})
}
func (dc *OriginDiscoveryClient) regServiceDiscover(nodeId string){
func (dc *OriginDiscoveryClient) regServiceDiscover(nodeId string) {
if nodeId == cluster.GetLocalNodeInfo().NodeId {
return
}
@@ -521,14 +518,14 @@ func (dc *OriginDiscoveryClient) regServiceDiscover(nodeId string){
req.NodeInfo.NodeId = cluster.localNodeInfo.NodeId
req.NodeInfo.ListenAddr = cluster.localNodeInfo.ListenAddr
req.NodeInfo.MaxRpcParamLen = cluster.localNodeInfo.MaxRpcParamLen
req.NodeInfo.PublicServiceList = cluster.localNodeInfo.PublicServiceList
req.NodeInfo.PublicServiceList = cluster.localNodeInfo.PublicServiceList
req.NodeInfo.Retire = dc.bRetire
req.NodeInfo.Private = cluster.localNodeInfo.Private
log.Debug("regServiceDiscover",log.String("nodeId",nodeId))
log.Debug("regServiceDiscover", log.String("nodeId", nodeId))
//向Master服务同步本Node服务信息
_,err := dc.AsyncCallNodeWithTimeout(3*time.Second,nodeId, RegServiceDiscover, &req, func(res *rpc.SubscribeDiscoverNotify, err error) {
_, err := dc.AsyncCallNodeWithTimeout(3*time.Second, nodeId, RegServiceDiscover, &req, func(res *rpc.SubscribeDiscoverNotify, err error) {
if err != nil {
log.Error("call "+RegServiceDiscover+" is fail :"+ err.Error())
log.Error("call " + RegServiceDiscover + " is fail :" + err.Error())
dc.tryRegServiceDiscover(nodeId)
return
}
@@ -538,12 +535,12 @@ func (dc *OriginDiscoveryClient) regServiceDiscover(nodeId string){
})
if err != nil {
log.Error("call "+ RegServiceDiscover+" is fail :"+ err.Error())
log.Error("call " + RegServiceDiscover + " is fail :" + err.Error())
dc.tryRegServiceDiscover(nodeId)
}
}
func (dc *OriginDiscoveryClient) setNodeInfo(masterNodeId string,nodeInfo *rpc.NodeInfo) bool{
func (dc *OriginDiscoveryClient) setNodeInfo(masterNodeId string, nodeInfo *rpc.NodeInfo) bool {
if nodeInfo == nil || nodeInfo.Private == true || nodeInfo.NodeId == dc.localNodeId {
return false
}
@@ -551,8 +548,8 @@ func (dc *OriginDiscoveryClient) setNodeInfo(masterNodeId string,nodeInfo *rpc.N
//筛选关注的服务
var discoverServiceSlice = make([]string, 0, 24)
for _, pubService := range nodeInfo.PublicServiceList {
if cluster.CanDiscoveryService(masterNodeId,pubService) == true {
discoverServiceSlice = append(discoverServiceSlice,pubService)
if cluster.CanDiscoveryService(masterNodeId, pubService) == true {
discoverServiceSlice = append(discoverServiceSlice, pubService)
}
}
@@ -604,8 +601,8 @@ func (cls *Cluster) checkOriginDiscovery(localNodeId string) (bool, bool) {
}
func (cls *Cluster) AddDiscoveryService(serviceName string, bPublicService bool) {
addServiceList := append([]string{},serviceName)
cls.localNodeInfo.ServiceList = append(addServiceList,cls.localNodeInfo.ServiceList...)
addServiceList := append([]string{}, serviceName)
cls.localNodeInfo.ServiceList = append(addServiceList, cls.localNodeInfo.ServiceList...)
if bPublicService {
cls.localNodeInfo.PublicServiceList = append(cls.localNodeInfo.PublicServiceList, serviceName)
}
@@ -616,7 +613,6 @@ func (cls *Cluster) AddDiscoveryService(serviceName string, bPublicService bool)
cls.mapServiceNode[serviceName][cls.localNodeInfo.NodeId] = struct{}{}
}
func (cls *Cluster) IsOriginMasterDiscoveryNode(nodeId string) bool {
return cls.getOriginMasterDiscoveryNodeInfo(nodeId) != nil
}
@@ -635,12 +631,12 @@ func (cls *Cluster) getOriginMasterDiscoveryNodeInfo(nodeId string) *NodeInfo {
return nil
}
func (dc *OriginDiscoveryClient) OnNatsConnected(){
func (dc *OriginDiscoveryClient) OnNatsConnected() {
masterNodes := GetCluster().GetOriginDiscovery().MasterNodeList
for i:=0;i<len(masterNodes);i++ {
for i := 0; i < len(masterNodes); i++ {
dc.regServiceDiscover(masterNodes[i].NodeId)
}
}
func (dc *OriginDiscoveryClient) OnNatsDisconnect(){
}
func (dc *OriginDiscoveryClient) OnNatsDisconnect() {
}

169
cluster/parsecfg.go
View File

@@ -1,6 +1,7 @@
package cluster
import (
"errors"
"fmt"
"github.com/duanhf2012/origin/v2/log"
"github.com/duanhf2012/origin/v2/rpc"
@@ -9,25 +10,24 @@ import (
"path/filepath"
"strings"
"time"
"errors"
)
var json = jsoniter.ConfigCompatibleWithStandardLibrary
type EtcdList struct {
NetworkName []string
Endpoints []string
Endpoints []string
}
type EtcdDiscovery struct {
DialTimeoutMillisecond time.Duration
TTLSecond int64
TTLSecond int64
EtcdList []EtcdList
}
type OriginDiscovery struct {
TTLSecond int64
TTLSecond int64
MasterNodeList []NodeInfo
}
@@ -35,40 +35,39 @@ type DiscoveryType int
const (
InvalidType = 0
OriginType = 1
EtcdType = 2
OriginType = 1
EtcdType = 2
)
const MinTTL = 3
type DiscoveryInfo struct {
discoveryType DiscoveryType
Etcd *EtcdDiscovery //etcd
Origin *OriginDiscovery //orign
Etcd *EtcdDiscovery //etcd
Origin *OriginDiscovery //origin
}
type NatsConfig struct {
NatsUrl string
NatsUrl string
NoRandomize bool
}
type RpcMode struct {
Typ string `json:"Type"`
Nats NatsConfig
Typ string `json:"Type"`
Nats NatsConfig
}
type NodeInfoList struct {
RpcMode RpcMode
Discovery DiscoveryInfo
NodeList []NodeInfo
RpcMode RpcMode
Discovery DiscoveryInfo
NodeList []NodeInfo
}
func (d *DiscoveryInfo) getDiscoveryType() DiscoveryType{
func (d *DiscoveryInfo) getDiscoveryType() DiscoveryType {
return d.discoveryType
}
func (d *DiscoveryInfo) setDiscovery(discoveryInfo *DiscoveryInfo) error{
func (d *DiscoveryInfo) setDiscovery(discoveryInfo *DiscoveryInfo) error {
var err error
err = d.setOrigin(discoveryInfo.Origin)
if err != nil {
@@ -83,30 +82,30 @@ func (d *DiscoveryInfo) setDiscovery(discoveryInfo *DiscoveryInfo) error{
return nil
}
func (d *DiscoveryInfo) setEtcd(etcd *EtcdDiscovery) error{
func (d *DiscoveryInfo) setEtcd(etcd *EtcdDiscovery) error {
if etcd == nil {
return nil
}
if d.discoveryType != InvalidType {
return fmt.Errorf("Repeat configuration of Discovery")
if d.discoveryType != InvalidType {
return fmt.Errorf("repeat configuration of Discovery")
}
//Endpoints不允许重复
mapAddr:=make (map[string]struct{})
mapAddr := make(map[string]struct{})
for _, n := range etcd.EtcdList {
for _,endPoint := range n.Endpoints {
if _,ok:=mapAddr[endPoint];ok == true {
return fmt.Errorf("etcd discovery config Etcd.EtcdList.Endpoints %+v is repeat",endPoint)
for _, endPoint := range n.Endpoints {
if _, ok := mapAddr[endPoint]; ok == true {
return fmt.Errorf("etcd discovery config Etcd.EtcdList.Endpoints %+v is repeat", endPoint)
}
mapAddr[endPoint] = struct{}{}
}
//networkName不允许重复
mapNetworkName := make(map[string]struct{})
for _,netName := range n.NetworkName{
if _,ok := mapNetworkName[netName];ok == true {
return fmt.Errorf("etcd discovery config Etcd.EtcdList.NetworkName %+v is repeat",n.NetworkName)
for _, netName := range n.NetworkName {
if _, ok := mapNetworkName[netName]; ok == true {
return fmt.Errorf("etcd discovery config Etcd.EtcdList.NetworkName %+v is repeat", n.NetworkName)
}
mapNetworkName[netName] = struct{}{}
@@ -124,13 +123,13 @@ func (d *DiscoveryInfo) setEtcd(etcd *EtcdDiscovery) error{
return nil
}
func (d *DiscoveryInfo) setOrigin(originDiscovery *OriginDiscovery) error{
if originDiscovery== nil || len(originDiscovery.MasterNodeList)==0 {
func (d *DiscoveryInfo) setOrigin(originDiscovery *OriginDiscovery) error {
if originDiscovery == nil || len(originDiscovery.MasterNodeList) == 0 {
return nil
}
if d.discoveryType != InvalidType {
return fmt.Errorf("Repeat configuration of Discovery")
return fmt.Errorf("repeat configuration of Discovery")
}
mapListenAddr := make(map[string]struct{})
@@ -204,7 +203,7 @@ func (cls *Cluster) readServiceConfig(filepath string) (interface{}, map[string]
return GlobalCfg, serviceConfig, mapNodeService, nil
}
func (cls *Cluster) SetRpcMode(cfgRpcMode *RpcMode,rpcMode *RpcMode) error {
func (cls *Cluster) SetRpcMode(cfgRpcMode *RpcMode, rpcMode *RpcMode) error {
//忽略掉没有设置的配置
if cfgRpcMode.Typ == "" {
return nil
@@ -212,16 +211,16 @@ func (cls *Cluster) SetRpcMode(cfgRpcMode *RpcMode,rpcMode *RpcMode) error {
//不允许重复的配置Rpc模式
if cfgRpcMode.Typ != "" && rpcMode.Typ != ""{
if cfgRpcMode.Typ != "" && rpcMode.Typ != "" {
return errors.New("repeat config RpcMode")
}
//检查Typ是否合法
if cfgRpcMode.Typ!="Nats" && cfgRpcMode.Typ!="Default" {
if cfgRpcMode.Typ != "Nats" && cfgRpcMode.Typ != "Default" {
return fmt.Errorf("RpcMode %s is not support", rpcMode.Typ)
}
if cfgRpcMode.Typ == "Nats" && len(cfgRpcMode.Nats.NatsUrl)==0 {
if cfgRpcMode.Typ == "Nats" && len(cfgRpcMode.Nats.NatsUrl) == 0 {
return fmt.Errorf("nats rpc mode config NatsUrl is empty")
}
@@ -230,7 +229,7 @@ func (cls *Cluster) SetRpcMode(cfgRpcMode *RpcMode,rpcMode *RpcMode) error {
return nil
}
func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []NodeInfo,RpcMode, error) {
func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []NodeInfo, RpcMode, error) {
var nodeInfoList []NodeInfo
var discoveryInfo DiscoveryInfo
var rpcMode RpcMode
@@ -238,27 +237,26 @@ func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []Node
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := os.ReadDir(clusterCfgPath)
if err != nil {
return discoveryInfo, nil,rpcMode, fmt.Errorf("Read dir %s is fail :%+v", clusterCfgPath, err)
return discoveryInfo, nil, rpcMode, fmt.Errorf("read dir %s is fail :%+v", clusterCfgPath, err)
}
//读取任何文件,只读符合格式的配置,目录下的文件可以自定义分文件
for _, f := range fileInfoList {
if f.IsDir() == false {
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
fileNodeInfoList, rerr := cls.ReadClusterConfig(filePath)
if rerr != nil {
return discoveryInfo, nil,rpcMode, fmt.Errorf("read file path %s is error:%+v", filePath, rerr)
fileNodeInfoList, rErr := cls.ReadClusterConfig(filePath)
if rErr != nil {
return discoveryInfo, nil, rpcMode, fmt.Errorf("read file path %s is error:%+v", filePath, rErr)
}
err = cls.SetRpcMode(&fileNodeInfoList.RpcMode,&rpcMode)
err = cls.SetRpcMode(&fileNodeInfoList.RpcMode, &rpcMode)
if err != nil {
return discoveryInfo, nil,rpcMode, err
return discoveryInfo, nil, rpcMode, err
}
err = discoveryInfo.setDiscovery(&fileNodeInfoList.Discovery)
if err != nil {
return discoveryInfo,nil,rpcMode,err
return discoveryInfo, nil, rpcMode, err
}
for _, nodeInfo := range fileNodeInfoList.NodeList {
@@ -270,10 +268,10 @@ func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []Node
}
if nodeId != rpc.NodeIdNull && (len(nodeInfoList) != 1) {
return discoveryInfo, nil,rpcMode, fmt.Errorf("nodeid %s configuration error in NodeList", nodeId)
return discoveryInfo, nil, rpcMode, fmt.Errorf("nodeid %s configuration error in NodeList", nodeId)
}
for i, _ := range nodeInfoList {
for i := range nodeInfoList {
for j, s := range nodeInfoList[i].ServiceList {
//私有结点不加入到Public服务列表中
if strings.HasPrefix(s, "_") == false && nodeInfoList[i].Private == false {
@@ -284,17 +282,17 @@ func (cls *Cluster) readLocalClusterConfig(nodeId string) (DiscoveryInfo, []Node
}
}
return discoveryInfo, nodeInfoList, rpcMode,nil
return discoveryInfo, nodeInfoList, rpcMode, nil
}
func (cls *Cluster) readLocalService(localNodeId string) error {
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := os.ReadDir(clusterCfgPath)
if err != nil {
return fmt.Errorf("Read dir %s is fail :%+v", clusterCfgPath, err)
return fmt.Errorf("read dir %s is fail :%+v", clusterCfgPath, err)
}
var globalCfg interface{}
var globalCfg interface{}
publicService := map[string]interface{}{}
nodeService := map[string]interface{}{}
@@ -304,12 +302,12 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
continue
}
if filepath.Ext(f.Name())!= ".json" {
if filepath.Ext(f.Name()) != ".json" {
continue
}
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
currGlobalCfg, serviceConfig, mapNodeService, err := cls.readServiceConfig(filePath)
currGlobalCfg, serviceConfig, mapNodeService, err := cls.readServiceConfig(filePath)
if err != nil {
continue
}
@@ -317,7 +315,7 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
if currGlobalCfg != nil {
//不允许重复的配置global配置
if globalCfg != nil {
return fmt.Errorf("[Global] does not allow repeated configuration in %s.",f.Name())
return fmt.Errorf("[Global] does not allow repeated configuration in %s", f.Name())
}
globalCfg = currGlobalCfg
}
@@ -325,21 +323,21 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
//保存公共配置
for _, s := range cls.localNodeInfo.ServiceList {
for {
splitServiceName := strings.Split(s,":")
splitServiceName := strings.Split(s, ":")
if len(splitServiceName) == 2 {
s = splitServiceName[0]
}
//取公共服务配置
pubCfg, ok := serviceConfig[s]
if ok == true {
if _,publicOk := publicService[s];publicOk == true {
return fmt.Errorf("public service [%s] does not allow repeated configuration in %s.",s,f.Name())
if _, publicOk := publicService[s]; publicOk == true {
return fmt.Errorf("public service [%s] does not allow repeated configuration in %s", s, f.Name())
}
publicService[s] = pubCfg
}
//取指定结点配置的服务
nodeServiceCfg,ok := mapNodeService[localNodeId]
nodeServiceCfg, ok := mapNodeService[localNodeId]
if ok == false {
break
}
@@ -348,8 +346,8 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
break
}
if _,nodeOK := nodeService[s];nodeOK == true {
return fmt.Errorf("NodeService NodeId[%d] Service[%s] does not allow repeated configuration in %s.",cls.localNodeInfo.NodeId,s,f.Name())
if _, nodeOK := nodeService[s]; nodeOK == true {
return fmt.Errorf("NodeService NodeId[%s] Service[%s] does not allow repeated configuration in %s", cls.localNodeInfo.NodeId, s, f.Name())
}
nodeService[s] = nodeCfg
break
@@ -359,7 +357,7 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
//组合所有的配置
for _, s := range cls.localNodeInfo.ServiceList {
splitServiceName := strings.Split(s,":")
splitServiceName := strings.Split(s, ":")
if len(splitServiceName) == 2 {
s = splitServiceName[0]
}
@@ -367,16 +365,16 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
//先从NodeService中找
var serviceCfg interface{}
var ok bool
serviceCfg,ok = nodeService[s]
serviceCfg, ok = nodeService[s]
if ok == true {
cls.localServiceCfg[s] =serviceCfg
cls.localServiceCfg[s] = serviceCfg
continue
}
//如果找不到从PublicService中找
serviceCfg,ok = publicService[s]
serviceCfg, ok = publicService[s]
if ok == true {
cls.localServiceCfg[s] =serviceCfg
cls.localServiceCfg[s] = serviceCfg
}
}
cls.globalCfg = globalCfg
@@ -387,23 +385,22 @@ func (cls *Cluster) readLocalService(localNodeId string) error {
func (cls *Cluster) parseLocalCfg() {
rpcInfo := NodeRpcInfo{}
rpcInfo.nodeInfo = cls.localNodeInfo
rpcInfo.client = rpc.NewLClient(rpcInfo.nodeInfo.NodeId,&cls.callSet)
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,":")
splitServiceName := strings.Split(serviceName, ":")
if len(splitServiceName) == 2 {
serviceName = splitServiceName[0]
templateServiceName := splitServiceName[1]
//记录模板
if _, ok := cls.mapTemplateServiceNode[templateServiceName]; ok == false {
cls.mapTemplateServiceNode[templateServiceName]=map[string]struct{}{}
cls.mapTemplateServiceNode[templateServiceName] = map[string]struct{}{}
}
cls.mapTemplateServiceNode[templateServiceName][serviceName] = struct{}{}
}
if _, ok := cls.mapServiceNode[serviceName]; ok == false {
cls.mapServiceNode[serviceName] = make(map[string]struct{})
}
@@ -427,7 +424,7 @@ func (cls *Cluster) InitCfg(localNodeId string) error {
cls.mapTemplateServiceNode = map[string]map[string]struct{}{}
//加载本地结点的NodeList配置
discoveryInfo, nodeInfoList,rpcMode, err := cls.readLocalClusterConfig(localNodeId)
discoveryInfo, nodeInfoList, rpcMode, err := cls.readLocalClusterConfig(localNodeId)
if err != nil {
return err
}
@@ -463,24 +460,24 @@ func (cls *Cluster) GetNodeIdByTemplateService(templateServiceName string, rpcCl
defer cls.locker.RUnlock()
mapServiceName := cls.mapTemplateServiceNode[templateServiceName]
for serviceName := range mapServiceName {
mapNodeId, ok := cls.mapServiceNode[serviceName]
if ok == true {
for nodeId, _ := range mapNodeId {
pClient,retire := GetCluster().getRpcClient(nodeId)
if pClient == nil || pClient.IsConnected() == false {
continue
}
for serviceName := range mapServiceName {
mapNodeId, ok := cls.mapServiceNode[serviceName]
if ok == true {
for nodeId := range mapNodeId {
pClient, retire := GetCluster().getRpcClient(nodeId)
if pClient == nil || pClient.IsConnected() == false {
continue
}
//如果需要筛选掉退休的，对retire状态的结点略过
if filterRetire == true && retire == true {
continue
}
//如果需要筛选掉退休的，对retire状态的结点略过
if filterRetire == true && retire == true {
continue
}
rpcClientList = append(rpcClientList,pClient)
}
}
}
rpcClientList = append(rpcClientList, pClient)
}
}
}
return nil, rpcClientList
}
@@ -490,8 +487,8 @@ func (cls *Cluster) GetNodeIdByService(serviceName string, rpcClientList []*rpc.
defer cls.locker.RUnlock()
mapNodeId, ok := cls.mapServiceNode[serviceName]
if ok == true {
for nodeId, _ := range mapNodeId {
pClient,retire := GetCluster().getRpcClient(nodeId)
for nodeId := range mapNodeId {
pClient, retire := GetCluster().getRpcClient(nodeId)
if pClient == nil || pClient.IsConnected() == false {
continue
}
@@ -501,7 +498,7 @@ func (cls *Cluster) GetNodeIdByService(serviceName string, rpcClientList []*rpc.
continue
}
rpcClientList = append(rpcClientList,pClient)
rpcClientList = append(rpcClientList, pClient)
}
}