package cluster import ( "fmt" "net" "os" "strconv" "strings" "time" "github.com/duanhf2012/origin/sysmodule" "github.com/duanhf2012/origin/service" "github.com/duanhf2012/origin/rpc" ) type RpcClient struct { nodeid int pclient *rpc.Client serverAddr string isConnect bool } type CCluster struct { port int cfg *ClusterConfig nodeclient map[int]*RpcClient reader net.Conn writer net.Conn LocalRpcClient *rpc.Client } func (slf *CCluster) ReadNodeInfo(nodeid int) error { var err error slf.cfg, err = ReadCfg("./config/cluster.json", nodeid) if err != nil { fmt.Printf("%v", err) return nil } return nil } func (slf *CCluster) GetClusterClient(id int) *rpc.Client { v, ok := slf.nodeclient[id] if ok == false { return nil } return v.pclient } func (slf *CCluster) GetBindUrl() string { return slf.cfg.currentNode.ServerAddr } func (slf *CCluster) AcceptRpc(tpcListen *net.TCPListener) error { for { conn, err := tpcListen.Accept() if err != nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "tpcListen.Accept error:%v", err) return err } go rpc.ServeConn(conn) } return nil } func (slf *CCluster) ListenService() error { bindStr := slf.GetBindUrl() tcpaddr, err := net.ResolveTCPAddr("tcp4", bindStr) if err != nil { service.GetLogger().Printf(sysmodule.LEVER_FATAL, "ResolveTCPAddr error:%v", err) os.Exit(1) return err } tcplisten, err2 := net.ListenTCP("tcp", tcpaddr) if err2 != nil { service.GetLogger().Printf(sysmodule.LEVER_FATAL, "ListenTCP error:%v", err) os.Exit(1) return err2 } slf.reader, slf.writer = net.Pipe() go rpc.ServeConn(slf.reader) slf.LocalRpcClient = rpc.NewClient(slf.writer) go slf.AcceptRpc(tcplisten) return nil } type CPing struct { TimeStamp int64 } type CPong struct { TimeStamp int64 } func (slf *CPing) Ping(ping *CPing, pong *CPong) error { pong.TimeStamp = ping.TimeStamp return nil } func (slf *CCluster) ConnService() error { ping := CPing{0} pong := CPong{0} rpc.RegisterName("CPing", "", &ping) //连接集群服务器 for _, nodeList := range slf.cfg.mapClusterNodeService { for _, node := range nodeList { slf.nodeclient[node.NodeID] = &RpcClient{node.NodeID, nil, node.ServerAddr, false} } } for { for _, rpcClient := range slf.nodeclient { //连接状态发送ping if rpcClient.isConnect == true { ping.TimeStamp = 0 err := rpcClient.pclient.Call("CPing.Ping", &ping, &pong) if err != nil { rpcClient.pclient.Close() rpcClient.pclient = nil rpcClient.isConnect = false continue } continue } //非连接状态重新连接 if rpcClient.pclient != nil { rpcClient.pclient.Close() rpcClient.pclient = nil } client, err := rpc.Dial("tcp", rpcClient.serverAddr) if err != nil { service.GetLogger().Printf(sysmodule.LEVER_WARN, "Connect nodeid:%d,address:%s fail", rpcClient.nodeid, rpcClient.serverAddr) continue } service.GetLogger().Printf(sysmodule.LEVER_INFO, "Connect nodeid:%d,address:%s succ", rpcClient.nodeid, rpcClient.serverAddr) v, _ := slf.nodeclient[rpcClient.nodeid] v.pclient = client v.isConnect = true } time.Sleep(time.Second * 2) } return nil } func (slf *CCluster) Init() error { if len(os.Args) < 2 { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Init error,param not find NodeId=number") return fmt.Errorf("param error not find NodeId=number") } parts := strings.Split(os.Args[1], "=") if len(parts) < 2 { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Init error,param not find NodeId=number") return fmt.Errorf("param error not find NodeId=number") } if parts[0] != "NodeId" { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Init error,param error not find NodeId=number") return fmt.Errorf("param error not find NodeId=number") } slf.nodeclient = make(map[int]*RpcClient) //读取配置 ret, err := strconv.Atoi(parts[1]) if err != nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Init parts[1] error,%v", err) return err } return slf.ReadNodeInfo(ret) } func (slf *CCluster) Start() error { service.InstanceServiceMgr().FetchService(slf.OnFetchService) //监听服务 slf.ListenService() //集群 go slf.ConnService() return nil } //Node.servicename.methodname //servicename.methodname //_servicename.methodname func (slf *CCluster) Call(NodeServiceMethod string, args interface{}, reply interface{}) error { var callServiceName string nodeidList := slf.GetNodeList(NodeServiceMethod, &callServiceName) if len(nodeidList) > 1 || len(nodeidList) < 1 { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Call(%s) not find nodes.", NodeServiceMethod) return fmt.Errorf("CCluster.Call(%s) not find nodes.", NodeServiceMethod) } nodeid := nodeidList[0] if nodeid == GetNodeId() { return slf.LocalRpcClient.Call(callServiceName, args, reply) } else { pclient := slf.GetClusterClient(nodeid) if pclient == nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Call(%s): NodeId %d is not find.", NodeServiceMethod, nodeid) return fmt.Errorf("CCluster.Call(%s): NodeId %d is not find.", NodeServiceMethod, nodeid) } err := pclient.Call(callServiceName, args, reply) if err != nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Call(%s) is fail:%v.", callServiceName, err) } return err } service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Call(%s) fail.", NodeServiceMethod) return fmt.Errorf("CCluster.Call(%s) fail.", NodeServiceMethod) } func (slf *CCluster) GetNodeList(NodeServiceMethod string, rpcServerMethod *string) []int { var nodename string var servicename string var methodname string var nodeidList []int parts := strings.Split(NodeServiceMethod, ".") if len(parts) == 2 { servicename = parts[0] methodname = parts[1] } else if len(parts) == 3 { nodename = parts[0] servicename = parts[1] methodname = parts[2] } else { return nodeidList } if nodename == "" { nodeidList = make([]int, 0) if servicename[:1] == "_" { servicename = servicename[1:] nodeidList = append(nodeidList, GetNodeId()) } else { nodeidList = slf.cfg.GetIdByService(servicename) } } else { nodeidList = slf.GetIdByNodeService(nodename, servicename) } *rpcServerMethod = servicename + "." + methodname return nodeidList } func (slf *CCluster) Go(bCast bool, NodeServiceMethod string, args interface{}) error { var callServiceName string nodeidList := slf.GetNodeList(NodeServiceMethod, &callServiceName) if len(nodeidList) < 1 { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Go(%s) not find nodes.", NodeServiceMethod) return fmt.Errorf("CCluster.Go(%s) not find nodes.", NodeServiceMethod) } if bCast == false && len(nodeidList) > 1 { return fmt.Errorf("CCluster.Go(%s) find more nodes", NodeServiceMethod) } for _, nodeid := range nodeidList { if nodeid == GetNodeId() { replyCall := slf.LocalRpcClient.Go(callServiceName, args, nil, nil) if replyCall.Error != nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Go(%s) fail:%v.", NodeServiceMethod, replyCall.Error) } return replyCall.Error } else { pclient := slf.GetClusterClient(nodeid) if pclient == nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Go(%s) NodeId %d not find client", NodeServiceMethod, nodeid) return fmt.Errorf("CCluster.Go(%s) NodeId %d not find client", NodeServiceMethod, nodeid) } replyCall := pclient.Go(callServiceName, args, nil, nil) if replyCall.Error != nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.Go(%s) fail:%v.", NodeServiceMethod, replyCall.Error) } return replyCall.Error } } return nil } func (slf *CCluster) CallNode(nodeid int, servicemethod string, args interface{}, reply interface{}) error { pclient := slf.GetClusterClient(nodeid) if pclient == nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.CallNode(%d,%s) NodeId not find client", nodeid, servicemethod) return fmt.Errorf("Call: NodeId %d is not find.", nodeid) } err := pclient.Call(servicemethod, args, reply) if err != nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.CallNode(%d,%s) fail:%v", nodeid, servicemethod, err) } return err } func (slf *CCluster) GoNode(nodeid int, args interface{}, servicemethod string) error { pclient := slf.GetClusterClient(nodeid) if pclient == nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.GoNode(%d,%s) NodeId not find client", nodeid, servicemethod) return fmt.Errorf("CCluster.GoNode(%d,%s) NodeId not find client", nodeid, servicemethod) } replyCall := pclient.Go(servicemethod, args, nil, nil) if replyCall.Error != nil { service.GetLogger().Printf(sysmodule.LEVER_ERROR, "CCluster.GoNode(%d,%s) fail:%v", nodeid, servicemethod, replyCall.Error) } return replyCall.Error } func (ws *CCluster) OnFetchService(iservice service.IService) error { rpc.RegisterName(iservice.GetServiceName(), "RPC_", iservice) return nil } //向远程服务器调用 //Node.servicename.methodname //servicename.methodname func Call(NodeServiceMethod string, args interface{}, reply interface{}) error { return InstanceClusterMgr().Call(NodeServiceMethod, args, reply) } func Go(NodeServiceMethod string, args interface{}) error { return InstanceClusterMgr().Go(false, NodeServiceMethod, args) } func CallNode(NodeId int, servicemethod string, args interface{}, reply interface{}) error { return InstanceClusterMgr().CallNode(NodeId, servicemethod, args, reply) } func GoNode(NodeId int, servicemethod string, args interface{}) error { return InstanceClusterMgr().GoNode(NodeId, args, servicemethod) } func CastGo(NodeServiceMethod string, args interface{}) error { return InstanceClusterMgr().Go(true, NodeServiceMethod, args) } var _self *CCluster func InstanceClusterMgr() *CCluster { if _self == nil { _self = new(CCluster) return _self } return _self } func (slf *CCluster) GetIdByNodeService(NodeName string, serviceName string) []int { return slf.cfg.GetIdByNodeService(NodeName, serviceName) } func (slf *CCluster) HasLocalService(serviceName string) bool { return slf.cfg.HasLocalService(serviceName) } func GetNodeId() int { return _self.cfg.currentNode.NodeID }