up

proxy/load_balance/config.go

@@ -30,7 +30,7 @@ var(
 func Init(reg registry.Registry) *LoadBalanceEtcdConf{
 func Init(reg registry.Registry) *LoadBalanceEtcdConf{
 	once.Do(func() {
 	once.Do(func() {
 		LoadBalanceConfig  = &LoadBalanceEtcdConf{}
 		LoadBalanceConfig  = &LoadBalanceEtcdConf{}
-		rb := LoadBanlanceFactory(LbRoundRobin)
+		rb := LoadBanlanceFactory(LbRandom)
 		//添加负载均衡器到配置中
 		//添加负载均衡器到配置中
 		LoadBalanceConfig.Attach(rb)
 		LoadBalanceConfig.Attach(rb)
 		LoadBalanceConfig.registry = reg
 		LoadBalanceConfig.registry = reg

proxy/load_balance/random.go

@@ -2,13 +2,17 @@ package load_balance
 
 
 import (
 import (
 	"errors"
 	"errors"
+	"fmt"
 	"github.com/syyongx/php2go"
 	"github.com/syyongx/php2go"
 	"math/rand"
 	"math/rand"
+	"strconv"
+	"sync"
 )
 )
 
 
 type RandomBalance struct {
 type RandomBalance struct {
 	curIndex map[string]int
 	curIndex map[string]int
 	rss      map[string][]string
 	rss      map[string][]string
+	lock               sync.Mutex
 }
 }
 
 
 func (r *RandomBalance) Add(serviceName string,params ...string) error {
 func (r *RandomBalance) Add(serviceName string,params ...string) error {
@@ -30,6 +34,11 @@ func (r *RandomBalance) Add(serviceName string,params ...string) error {
 }
 }
 
 
 func (r *RandomBalance) Next(key string) string {
 func (r *RandomBalance) Next(key string) string {
+	defer func() {
+		if r := recover(); r != nil {
+			return
+		}
+	}()
 	if r.rss == nil || len(r.rss) == 0 {
 	if r.rss == nil || len(r.rss) == 0 {
 		return ""
 		return ""
 	}
 	}
@@ -55,5 +64,50 @@ func (r *RandomBalance) Get(key ...string) (string, error) {
 
 
 
 
 func (r *RandomBalance) Update() {
 func (r *RandomBalance) Update() {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	fmt.Println("更新负载均衡配置.....")
+	allServiceInfo := LoadBalanceConfig.GetLoadBalanceList()
+	if allServiceInfo == nil || allServiceInfo.ServiceMap == nil{
+		return
+	}
+
+	//删除不存在的服务节点信息
+	for serviceName,_ := range r.rss{
+		if _,ok :=  allServiceInfo.ServiceMap[serviceName];!ok{
+			delete(r.rss,serviceName)
+			delete(r.curIndex,serviceName)
+			continue
+		}
+		//循环etcd中的节点信息
+		var tmpNodes []string
+		for _,etcdServiceNode := range allServiceInfo.ServiceMap[serviceName].Nodes{
+			if etcdServiceNode.IP == ""{
+				continue
+			}
+			if etcdServiceNode.Port == 0 {
+				tmpNodes = append(tmpNodes,etcdServiceNode.IP)
+			}else{
+				tmpNodes = append(tmpNodes,fmt.Sprintf("%s:%s",etcdServiceNode.IP,strconv.Itoa(etcdServiceNode.Port)))
+			}
+		}
+		r.rss[serviceName] = tmpNodes
+	}
+
+
+	for _,service := range allServiceInfo.ServiceMap{
+		serviceName := service.Name
+		nodes := service.Nodes
+		for _,node :=range  nodes{
+			if node.IP == ""{
+				continue
+			}
+			if node.Port == 0{
+				r.Add(serviceName,node.IP)
+			}else{
+				r.Add(serviceName,fmt.Sprintf("%s:%s",node.IP,strconv.Itoa(node.Port)))
+			}
+		}
+	}
 
 
 }
 }

proxy/load_balance/round_robin.go

@@ -41,7 +41,6 @@ func (r *RoundRobinBalance) Add(serviceName string,params ...string) error {
 func (r *RoundRobinBalance) Next(key string) string {
 func (r *RoundRobinBalance) Next(key string) string {
 	defer func() {
 	defer func() {
 		if r := recover(); r != nil {
 		if r := recover(); r != nil {
-			fmt.Println("````````````````````")
 			return
 			return
 		}
 		}
 	}()
 	}()
@@ -66,7 +65,7 @@ func (r *RoundRobinBalance) Next(key string) string {
 	if r.curIndex[key] >= lens {
 	if r.curIndex[key] >= lens {
 		r.curIndex[key] = 0
 		r.curIndex[key] = 0
 	}
 	}
-	php2go.array_
+
 	curAddr := r.rss[key][r.curIndex[key]]
 	curAddr := r.rss[key][r.curIndex[key]]
 	r.curIndex[key] = (r.curIndex[key] + 1) % lens
 	r.curIndex[key] = (r.curIndex[key] + 1) % lens
 	return curAddr
 	return curAddr
@@ -93,6 +92,7 @@ func (r *RoundRobinBalance) Update() {
 	for serviceName,_ := range r.rss{
 	for serviceName,_ := range r.rss{
 		if _,ok :=  allServiceInfo.ServiceMap[serviceName];!ok{
 		if _,ok :=  allServiceInfo.ServiceMap[serviceName];!ok{
 			delete(r.rss,serviceName)
 			delete(r.rss,serviceName)
+			delete(r.curIndex,serviceName)
 			continue
 			continue
 		}
 		}
 		//循环etcd中的节点信息
 		//循环etcd中的节点信息