一致性hash优化 (#320)

* 一致性hash优化：
同一服务有多个obj的情况
同一hash值调用不同的obj会hash到不同的服务器
因为ChMap.Add会根据String(): Proto -h Host -p Port -t Timeout -d Container"计算hash,导致顺序不一致

* ChMap 中 Remove 方法实现对标 Add

* 升级 opentracing, 解决 opentracing 创建编译不通过

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/golang/protobuf v1.4.2
 	github.com/opentracing-contrib/go-observer v0.0.0-20170622124052-a52f23424492 // indirect
 	github.com/opentracing/opentracing-go v1.1.0
-	github.com/openzipkin-contrib/zipkin-go-opentracing v0.3.5
+	github.com/openzipkin-contrib/zipkin-go-opentracing v0.4.5
 	github.com/openzipkin/zipkin-go v0.2.2
 	github.com/stretchr/testify v1.6.1
 )

tars/endpointmanager.go

@@ -249,7 +249,7 @@ func (e *tarsEndpointManager) checkStatus() {
 				}
 				e.epLock.Unlock()
 
-				e.activeEpHashMap.Remove(ep.Key)
+				e.activeEpHashMap.Remove(ep)
 			}
 
 			if needCheck {

tars/util/consistenthash/consistenthash.go

@@ -19,7 +19,7 @@ type ChMap struct {
 
 // KV is the key value type.
 type KV interface {
-	String() string
+	HashKey() string
 }
 
 // NewChMap  create a ChMap which has replicates of virtual nodes.
@@ -71,32 +71,32 @@ func (c *ChMap) FindUint32(key uint32) (KV, bool) {
 func (c *ChMap) Add(node KV) error {
 	c.lock.Lock()
 	defer c.lock.Unlock()
-	if _, ok := c.mapValues[node.String()]; ok {
+	if _, ok := c.mapValues[node.HashKey()]; ok {
 		return errors.New("node already exists")
 	}
 	for i := 0; i < c.replicates; i++ {
-		virtualHost := fmt.Sprintf("%d#%s", i, node.String())
+		virtualHost := fmt.Sprintf("%d#%s", i, node.HashKey())
 		virtualKey := crc32.ChecksumIEEE([]byte(virtualHost))
 		c.hashRing[virtualKey] = node
 		c.sortedKeys = append(c.sortedKeys, virtualKey)
 	}
 	sort.Slice(c.sortedKeys, func(x int, y int) bool {
 		return c.sortedKeys[x] < c.sortedKeys[y]
 	})
-	c.mapValues[node.String()] = true
+	c.mapValues[node.HashKey()] = true
 	return nil
 }
 
 // Remove remove the node and all the vatual nodes from the key
-func (c *ChMap) Remove(node string) error {
+func (c *ChMap) Remove(node KV) error {
 	c.lock.Lock()
 	defer c.lock.Unlock()
-	if _, ok := c.mapValues[node]; !ok {
+	if _, ok := c.mapValues[node.HashKey()]; !ok {
 		return errors.New("host already removed")
 	}
-	delete(c.mapValues, node)
+	delete(c.mapValues, node.HashKey())
 	for i := 0; i < c.replicates; i++ {
-		virtualHost := fmt.Sprintf("%d#%s", i, node)
+		virtualHost := fmt.Sprintf("%d#%s", i, node.HashKey())
 		virtualKey := crc32.ChecksumIEEE([]byte(virtualHost))
 		delete(c.hashRing, virtualKey)
 	}

tars/util/endpoint/endpoint.go

@@ -26,6 +26,10 @@ func (e Endpoint) String() string {
 	return fmt.Sprintf("%s -h %s -p %d -t %d -d %s", e.Proto, e.Host, e.Port, e.Timeout, e.Container)
 }
 
+func (e Endpoint) HashKey() string {
+	return e.Host
+}
+
 func (e Endpoint) IsTcp() bool {
 	return e.Istcp == TCP || e.Istcp == SSL
 }