Ma/origin
1
0
Fork 0
mirror of https://github.com/duanhf2012/origin.git synced 2026-02-03 22:45:13 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

新增全局配置

Browse Source
This commit is contained in:
orgin
2022-04-02 14:02:33 +08:00
parent 63199bf862
commit 68b891df51
2 changed files with 80 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
cluster/cluster.go
View File

@@ -41,8 +41,9 @@ type NodeRpcInfo struct {
var cluster Cluster
type Cluster struct {
localNodeInfo NodeInfo //本结点配置信息
masterDiscoveryNodeList []NodeInfo //配置发现Master结点
localNodeInfo NodeInfo //本结点配置信息
masterDiscoveryNodeList []NodeInfo //配置发现Master结点
globalCfg interface{} //全局配置
localServiceCfg map[string]interface{} //map[serviceName]配置数据*
mapRpc map[int]NodeRpcInfo //nodeId
@@ -418,3 +419,7 @@ func HasService(nodeId int, serviceName string) bool {
return false
}
func (cls *Cluster) GetGlobalCfg() interface{} {
return cls.globalCfg
}

142
cluster/parsecfg.go
View File

@@ -10,12 +10,13 @@ import (
)
var json = jsoniter.ConfigCompatibleWithStandardLibrary
type NodeInfoList struct {
MasterDiscoveryNode []NodeInfo //用于服务发现Node
NodeList []NodeInfo
MasterDiscoveryNode []NodeInfo //用于服务发现Node
NodeList []NodeInfo
}
func (cls *Cluster) ReadClusterConfig(filepath string) (*NodeInfoList,error) {
func (cls *Cluster) ReadClusterConfig(filepath string) (*NodeInfoList, error) {
c := &NodeInfoList{}
d, err := ioutil.ReadFile(filepath)
if err != nil {
@@ -26,119 +27,123 @@ func (cls *Cluster) ReadClusterConfig(filepath string) (*NodeInfoList,error) {
return nil, err
}
return c,nil
return c, nil
}
func (cls *Cluster) readServiceConfig(filepath string) (map[string]interface{},map[int]map[string]interface{},error) {
func (cls *Cluster) readServiceConfig(filepath string) (interface{}, map[string]interface{}, map[int]map[string]interface{}, error) {
c := map[string]interface{}{}
//读取配置
d, err := ioutil.ReadFile(filepath)
if err != nil {
return nil,nil, err
return nil, nil, nil, err
}
err = json.Unmarshal(d, &c)
if err != nil {
return nil,nil, err
return nil, nil, nil, err
}
GlobalCfg, ok := c["Global"]
serviceConfig := map[string]interface{}{}
serviceCfg,ok := c["Service"]
serviceCfg, ok := c["Service"]
if ok == true {
serviceConfig = serviceCfg.(map[string]interface{})
}
mapNodeService := map[int]map[string]interface{}{}
nodeServiceCfg,ok := c["NodeService"]
nodeServiceCfg, ok := c["NodeService"]
if ok == true {
nodeServiceList := nodeServiceCfg.([]interface{})
for _,v := range nodeServiceList{
serviceCfg :=v.(map[string]interface{})
nodeId,ok := serviceCfg["NodeId"]
for _, v := range nodeServiceList {
serviceCfg := v.(map[string]interface{})
nodeId, ok := serviceCfg["NodeId"]
if ok == false {
log.SFatal("NodeService list not find nodeId field")
}
mapNodeService[int(nodeId.(float64))] = serviceCfg
}
}
return serviceConfig,mapNodeService,nil
return GlobalCfg, serviceConfig, mapNodeService, nil
}
func (cls *Cluster) readLocalClusterConfig(nodeId int) ([]NodeInfo,[]NodeInfo,error) {
func (cls *Cluster) readLocalClusterConfig(nodeId int) ([]NodeInfo, []NodeInfo, error) {
var nodeInfoList []NodeInfo
var masterDiscoverNodeList []NodeInfo
clusterCfgPath :=strings.TrimRight(configDir,"/") +"/cluster"
fileInfoList,err := ioutil.ReadDir(clusterCfgPath)
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := ioutil.ReadDir(clusterCfgPath)
if err != nil {
return nil,nil,fmt.Errorf("Read dir %s is fail :%+v",clusterCfgPath,err)
return nil, nil, fmt.Errorf("Read dir %s is fail :%+v", clusterCfgPath, err)
}
//读取任何文件,只读符合格式的配置,目录下的文件可以自定义分文件
for _,f := range fileInfoList{
for _, f := range fileInfoList {
if f.IsDir() == false {
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath,"/"),"\\")+"/"+f.Name()
localNodeInfoList,err := cls.ReadClusterConfig(filePath)
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
localNodeInfoList, err := cls.ReadClusterConfig(filePath)
if err != nil {
return nil,nil,fmt.Errorf("read file path %s is error:%+v" ,filePath,err)
return nil, nil, fmt.Errorf("read file path %s is error:%+v", filePath, err)
}
masterDiscoverNodeList = append(masterDiscoverNodeList,localNodeInfoList.MasterDiscoveryNode...)
for _,nodeInfo := range localNodeInfoList.NodeList {
masterDiscoverNodeList = append(masterDiscoverNodeList, localNodeInfoList.MasterDiscoveryNode...)
for _, nodeInfo := range localNodeInfoList.NodeList {
if nodeInfo.NodeId == nodeId || nodeId == 0 {
nodeInfoList = append(nodeInfoList,nodeInfo)
nodeInfoList = append(nodeInfoList, nodeInfo)
}
}
}
}
if nodeId != 0 && (len(nodeInfoList)!=1){
return nil,nil,fmt.Errorf("%d configurations were found for the configuration with node ID %d!",len(nodeInfoList),nodeId)
if nodeId != 0 && (len(nodeInfoList) != 1) {
return nil, nil, fmt.Errorf("%d configurations were found for the configuration with node ID %d!", len(nodeInfoList), nodeId)
}
for i,_ := range nodeInfoList{
for j,s := range nodeInfoList[i].ServiceList{
for i, _ := range nodeInfoList {
for j, s := range nodeInfoList[i].ServiceList {
//私有结点不加入到Public服务列表中
if strings.HasPrefix(s,"_") == false && nodeInfoList[i].Private==false {
nodeInfoList[i].PublicServiceList = append(nodeInfoList[i].PublicServiceList,strings.TrimLeft(s,"_"))
}else{
nodeInfoList[i].ServiceList[j] = strings.TrimLeft(s,"_")
if strings.HasPrefix(s, "_") == false && nodeInfoList[i].Private == false {
nodeInfoList[i].PublicServiceList = append(nodeInfoList[i].PublicServiceList, strings.TrimLeft(s, "_"))
} else {
nodeInfoList[i].ServiceList[j] = strings.TrimLeft(s, "_")
}
}
}
return masterDiscoverNodeList,nodeInfoList,nil
return masterDiscoverNodeList, nodeInfoList, nil
}
func (cls *Cluster) readLocalService(localNodeId int) error {
clusterCfgPath :=strings.TrimRight(configDir,"/") +"/cluster"
fileInfoList,err := ioutil.ReadDir(clusterCfgPath)
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := ioutil.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)
}
//读取任何文件,只读符合格式的配置,目录下的文件可以自定义分文件
for _,f := range fileInfoList {
for _, f := range fileInfoList {
if f.IsDir() == false {
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
serviceConfig,mapNodeService,err := cls.readServiceConfig(filePath)
currGlobalCfg, serviceConfig, mapNodeService, err := cls.readServiceConfig(filePath)
if err != nil {
continue
}
for _,s := range cls.localNodeInfo.ServiceList{
for{
if currGlobalCfg != nil {
cls.globalCfg = currGlobalCfg
}
for _, s := range cls.localNodeInfo.ServiceList {
for {
//取公共服务配置
pubCfg,ok := serviceConfig[s]
pubCfg, ok := serviceConfig[s]
if ok == true {
cls.localServiceCfg[s] = pubCfg
}
//如果结点也配置了该服务，则覆盖之
nodeService,ok := mapNodeService[localNodeId]
nodeService, ok := mapNodeService[localNodeId]
if ok == false {
break
}
sCfg,ok := nodeService[s]
if ok == false{
sCfg, ok := nodeService[s]
if ok == false {
break
}
@@ -152,22 +157,21 @@ func (cls *Cluster) readLocalService(localNodeId int) error {
return nil
}
func (cls *Cluster) parseLocalCfg(){
func (cls *Cluster) parseLocalCfg() {
cls.mapIdNode[cls.localNodeInfo.NodeId] = cls.localNodeInfo
for _,sName := range cls.localNodeInfo.ServiceList{
if _,ok:=cls.mapServiceNode[sName];ok==false{
for _, sName := range cls.localNodeInfo.ServiceList {
if _, ok := cls.mapServiceNode[sName]; ok == false {
cls.mapServiceNode[sName] = make(map[int]struct{})
}
cls.mapServiceNode[sName][cls.localNodeInfo.NodeId]= struct{}{}
cls.mapServiceNode[sName][cls.localNodeInfo.NodeId] = struct{}{}
}
}
func (cls *Cluster) checkDiscoveryNodeList(discoverMasterNode []NodeInfo) bool{
for i:=0;i<len(discoverMasterNode)-1;i++{
for j:=i+1;j<len(discoverMasterNode);j++{
func (cls *Cluster) checkDiscoveryNodeList(discoverMasterNode []NodeInfo) bool {
for i := 0; i < len(discoverMasterNode)-1; i++ {
for j := i + 1; j < len(discoverMasterNode); j++ {
if discoverMasterNode[i].NodeId == discoverMasterNode[j].NodeId ||
discoverMasterNode[i].ListenAddr == discoverMasterNode[j].ListenAddr {
return false
@@ -178,19 +182,19 @@ func (cls *Cluster) checkDiscoveryNodeList(discoverMasterNode []NodeInfo) bool{
return true
}
func (cls *Cluster) InitCfg(localNodeId int) error{
func (cls *Cluster) InitCfg(localNodeId int) error {
cls.localServiceCfg = map[string]interface{}{}
cls.mapRpc = map[int] NodeRpcInfo{}
cls.mapRpc = map[int]NodeRpcInfo{}
cls.mapIdNode = map[int]NodeInfo{}
cls.mapServiceNode = map[string]map[int]struct{}{}
//加载本地结点的NodeList配置
discoveryNode,nodeInfoList,err := cls.readLocalClusterConfig(localNodeId)
discoveryNode, nodeInfoList, err := cls.readLocalClusterConfig(localNodeId)
if err != nil {
return err
}
cls.localNodeInfo = nodeInfoList[0]
if cls.checkDiscoveryNodeList(discoveryNode) ==false {
if cls.checkDiscoveryNodeList(discoveryNode) == false {
return fmt.Errorf("DiscoveryNode config is error!")
}
cls.masterDiscoveryNodeList = discoveryNode
@@ -209,39 +213,39 @@ func (cls *Cluster) InitCfg(localNodeId int) error{
func (cls *Cluster) IsConfigService(serviceName string) bool {
cls.locker.RLock()
defer cls.locker.RUnlock()
mapNode,ok := cls.mapServiceNode[serviceName]
mapNode, ok := cls.mapServiceNode[serviceName]
if ok == false {
return false
}
_,ok = mapNode[cls.localNodeInfo.NodeId]
_, ok = mapNode[cls.localNodeInfo.NodeId]
return ok
}
func (cls *Cluster) GetNodeIdByService(serviceName string,rpcClientList []*rpc.Client,bAll bool) (error,int) {
func (cls *Cluster) GetNodeIdByService(serviceName string, rpcClientList []*rpc.Client, bAll bool) (error, int) {
cls.locker.RLock()
defer cls.locker.RUnlock()
mapNodeId,ok := cls.mapServiceNode[serviceName]
mapNodeId, ok := cls.mapServiceNode[serviceName]
count := 0
if ok == true {
for nodeId,_ := range mapNodeId {
for nodeId, _ := range mapNodeId {
pClient := GetCluster().getRpcClient(nodeId)
if pClient==nil || (bAll == false && pClient.IsConnected()==false) {
if pClient == nil || (bAll == false && pClient.IsConnected() == false) {
continue
}
rpcClientList[count] = pClient
count++
if count>=cap(rpcClientList) {
if count >= cap(rpcClientList) {
break
}
}
}
return nil,count
return nil, count
}
func (cls *Cluster) getServiceCfg(serviceName string) interface{}{
v,ok := cls.localServiceCfg[serviceName]
func (cls *Cluster) getServiceCfg(serviceName string) interface{} {
v, ok := cls.localServiceCfg[serviceName]
if ok == false {
return nil
}
@@ -249,8 +253,8 @@ func (cls *Cluster) getServiceCfg(serviceName string) interface{}{
return v
}
func (cls *Cluster) GetServiceCfg(serviceName string) interface{}{
serviceCfg,ok := cls.localServiceCfg[serviceName]
func (cls *Cluster) GetServiceCfg(serviceName string) interface{} {
serviceCfg, ok := cls.localServiceCfg[serviceName]
if ok == false {
return nil
}