NewLeastActiveLoadBalance

cluster/loadbalance/least_active.go

@@ -14,19 +14,21 @@
 package loadbalance
 
 import (
-	"github.com/dubbo/go-for-apache-dubbo/cluster"
-	"github.com/dubbo/go-for-apache-dubbo/common/extension"
-	"github.com/dubbo/go-for-apache-dubbo/filter"
-	"github.com/dubbo/go-for-apache-dubbo/protocol"
 	"math/rand"
 )
 
+import (
+	"github.com/apache/dubbo-go/cluster"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/protocol"
+)
+
 const (
-	leastActive = "leastactive"
+	LeastActive = "leastactive"
 )
 
 func init() {
-	extension.SetLoadbalance(leastActive, NewLeastActiveLoadBalance)
+	extension.SetLoadbalance(LeastActive, NewLeastActiveLoadBalance)
 }
 
 type leastActiveLoadBalance struct {
@@ -37,9 +39,8 @@ func NewLeastActiveLoadBalance() cluster.LoadBalance {
 }
 
 func (lb *leastActiveLoadBalance) Select(invokers []protocol.Invoker, invocation protocol.Invocation) protocol.Invoker {
-
 	count := len(invokers)
-	if invokers == nil || count == 0 {
+	if count == 0 {
 		return nil
 	}
 	if count == 1 {
@@ -48,17 +49,17 @@ func (lb *leastActiveLoadBalance) Select(invokers []protocol.Invoker, invocation
 
 	var (
 		leastActive  int32 = -1                 // The least active value of all invokers
-		totalWeight  int64 = 0                  // The number of invokers having the same least active value (leastActive)
+		totalWeight  int64 = 0                  // The number of invokers having the same least active value (LEAST_ACTIVE)
 		firstWeight  int64 = 0                  // Initial value, used for comparision
-		leastIndexes       = make([]int, count) // The index of invokers having the same least active value (leastActive)
-		leastCount         = 0                  // The number of invokers having the same least active value (leastActive)
+		leastIndexes       = make([]int, count) // The index of invokers having the same least active value (LEAST_ACTIVE)
+		leastCount         = 0                  // The number of invokers having the same least active value (LEAST_ACTIVE)
 		sameWeight         = true               // Every invoker has the same weight value?
 	)
 
 	for i := 0; i < count; i++ {
 		invoker := invokers[i]
 		// Active number
-		active := filter.GetStatus(invoker.GetUrl(), invocation.MethodName()).GetActive()
+		active := protocol.GetStatus(invoker.GetUrl(), invocation.MethodName()).GetActive()
 		// current weight (maybe in warmUp)
 		weight := GetWeight(invoker, invocation)
 		// There are smaller active services

cluster/loadbalance/least_active_test.go

@@ -0,0 +1,67 @@
+package loadbalance
+
+import (
+	"context"
+	"fmt"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+func TestLeastActiveSelect(t *testing.T) {
+	loadBalance := NewLeastActiveLoadBalance()
+
+	var invokers []protocol.Invoker
+
+	url, _ := common.NewURL(context.TODO(), "dubbo://192.168.1.0:20000/org.apache.demo.HelloService")
+	invokers = append(invokers, protocol.NewBaseInvoker(url))
+	i := loadBalance.Select(invokers, &invocation.RPCInvocation{})
+	assert.True(t, i.GetUrl().URLEqual(url))
+
+	for i := 1; i < 10; i++ {
+		url, _ := common.NewURL(context.TODO(), fmt.Sprintf("dubbo://192.168.1.%v:20000/org.apache.demo.HelloService", i))
+		invokers = append(invokers, protocol.NewBaseInvoker(url))
+	}
+	loadBalance.Select(invokers, &invocation.RPCInvocation{})
+}
+
+func TestLeastActiveByWeight(t *testing.T) {
+	loadBalance := NewLeastActiveLoadBalance()
+
+	var invokers []protocol.Invoker
+	loop := 3
+	for i := 1; i <= loop; i++ {
+		url, _ := common.NewURL(context.TODO(), fmt.Sprintf("test%v://192.168.1.%v:20000/org.apache.demo.HelloService?weight=%v", i, i, i))
+		invokers = append(invokers, protocol.NewBaseInvoker(url))
+	}
+
+	inv := new(invocation.RPCInvocation)
+	inv.SetMethod("test")
+	protocol.BeginCount(invokers[2].GetUrl(), inv.MethodName())
+
+	loop = 10000
+
+	var (
+		firstCount  int
+		secondCount int
+	)
+
+	for i := 1; i <= loop; i++ {
+		invoker := loadBalance.Select(invokers, inv)
+		if invoker.GetUrl().Protocol == "test1" {
+			firstCount++
+		} else if invoker.GetUrl().Protocol == "test2" {
+			secondCount++
+		}
+	}
+
+	assert.Equal(t, firstCount+secondCount, loop)
+}

filter/impl/active_filter.go

@@ -14,13 +14,13 @@
 package impl
 
 import (
-	"github.com/dubbo/go-for-apache-dubbo/common/logger"
+	"github.com/apache/dubbo-go/common/logger"
 )
 
 import (
-	"github.com/dubbo/go-for-apache-dubbo/common/extension"
-	"github.com/dubbo/go-for-apache-dubbo/filter"
-	"github.com/dubbo/go-for-apache-dubbo/protocol"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/filter"
+	"github.com/apache/dubbo-go/protocol"
 )
 
 const active = "active"
@@ -35,13 +35,13 @@ type ActiveFilter struct {
 func (ef *ActiveFilter) Invoke(invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
 	logger.Infof("invoking active filter. %v,%v", invocation.MethodName(), len(invocation.Arguments()))
 
-	filter.BeginCount(invoker.GetUrl(), invocation.MethodName())
+	protocol.BeginCount(invoker.GetUrl(), invocation.MethodName())
 	return invoker.Invoke(invocation)
 }
 
 func (ef *ActiveFilter) OnResponse(result protocol.Result, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
 
-	filter.EndCount(invoker.GetUrl(), invocation.MethodName())
+	protocol.EndCount(invoker.GetUrl(), invocation.MethodName())
 	return result
 }
 

go.mod

@@ -1,6 +1,7 @@
 module github.com/apache/dubbo-go
 
 require (
+	github.com/dubbo/go-for-apache-dubbo v1.0.0
 	github.com/dubbogo/getty v1.0.7
 	github.com/dubbogo/hessian2 v1.0.1
 	github.com/pkg/errors v0.8.1

go.sum

@@ -1,8 +1,12 @@
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dubbo/go-for-apache-dubbo v1.0.0 h1:d1+EmiQNGjOZuEprh7ru/aa/5/pu3lvl8vBkdFEvLuw=
+github.com/dubbo/go-for-apache-dubbo v1.0.0/go.mod h1:seatOl29ahGPWuMz4+VqqB6BSNYdHLXtnpPJ54U988Q=
+github.com/dubbogo/getty v0.0.0-20190523180329-bdf5e640ea53/go.mod h1:cRMSuoCmwc5lULFFnYZTxyCfZhObmRTNbS7XRnPNHSo=
 github.com/dubbogo/getty v1.0.7 h1:5Hg+JwXyCKm9Yr4yJkm98ahhnoa8c2h6br5QJxwQ+YU=
 github.com/dubbogo/getty v1.0.7/go.mod h1:cRMSuoCmwc5lULFFnYZTxyCfZhObmRTNbS7XRnPNHSo=
+github.com/dubbogo/hessian2 v0.0.0-20190526221400-d5610bbd0a41/go.mod h1:XFGDn4oSZX26zkcfhkM/fCJrOqwQJxk/xgWW1KMJBKM=
 github.com/dubbogo/hessian2 v1.0.1 h1:ztI7gJxR3Isxrrl2jE1IZKX61eNR93eRKGhn49vPEX8=
 github.com/dubbogo/hessian2 v1.0.1/go.mod h1:XFGDn4oSZX26zkcfhkM/fCJrOqwQJxk/xgWW1KMJBKM=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=

filter/RpcStatus.go → protocol/RpcStatus.go

@@ -11,16 +11,19 @@
 // limitations under the License.
 
 // @author yiji@apache.org
-package filter
+package protocol
 
 import (
-	"github.com/dubbo/go-for-apache-dubbo/common"
 	"sync"
 	"sync/atomic"
 )
 
+import (
+	"github.com/apache/dubbo-go/common"
+)
+
 var (
-	methodStatistics = sync.Map{} // url -> { methodName : RpcStatus}
+	methodStatistics sync.Map // url -> { methodName : RpcStatus}
 )
 
 type RpcStatus struct {
@@ -32,22 +35,22 @@ func (rpc *RpcStatus) GetActive() int32 {
 }
 
 func GetStatus(url common.URL, methodName string) *RpcStatus {
-	identity := url.Key()
-	methodMap, found := methodStatistics.Load(identity)
+	identifier := url.Key()
+	methodMap, found := methodStatistics.Load(identifier)
 	if !found {
-		methodMap = sync.Map{}
-		methodStatistics.Store(identity, methodMap)
+		methodMap = &sync.Map{}
+		methodStatistics.Store(identifier, methodMap)
 	}
 
-	methodActive := methodMap.(sync.Map)
+	methodActive := methodMap.(*sync.Map)
 	rpcStatus, found := methodActive.Load(methodName)
 	if !found {
-		rpcStatus = RpcStatus{}
+		rpcStatus = &RpcStatus{}
 		methodActive.Store(methodName, rpcStatus)
 	}
 
-	status := rpcStatus.(RpcStatus)
-	return &status
+	status := rpcStatus.(*RpcStatus)
+	return status
 }
 
 func BeginCount(url common.URL, methodName string) {