Merge pull request #204 from BeyondWUXF/release/3.0

解决k8s动态扩缩容后一致性hash异常情况

servant/libservant/EndpointManager.cpp

@@ -758,6 +758,8 @@ void EndpointManager::updateEndpoints(const set<EndpointInfo> & active, const se
 
 	_activeProxys.clear();
 	_regProxys.clear();
+	_indexActiveProxys.clear();
+	_sortActivProxys.clear();
 
 	if(!active.empty())
 	{
@@ -797,6 +799,10 @@ void EndpointManager::updateEndpoints(const set<EndpointInfo> & active, const se
 
 		_regProxys.insert(make_pair(iter->cmpDesc(),iterAdapter->second));
 
+		const string &host = iterAdapter->second->endpoint().host();
+		_indexActiveProxys.insert(make_pair(host, iterAdapter->second));
+		_sortActivProxys.insert(make_pair(host, iterAdapter->second));
+
 		//设置该节点的静态权重值
 		iterAdapter->second->setWeight(iter->weight());
 	}
@@ -1010,6 +1016,7 @@ AdapterProxy* EndpointManager::getHashProxyForWeight(int64_t hashCode, bool bSta
         }
         else
         {
+            TLOGWARN("[EndpointManager::getHashProxyForWeight, hash not active," << _objectProxy->name() << "@" << _vRegProxys[iIndex]->endpoint().desc() << endl);
             if(_activeProxys.empty())
             {
                 TLOGERROR("[EndpointManager::getHashProxyForWeight _activeEndpoints is empty], bStatic:" << bStatic << endl);
@@ -1096,7 +1103,47 @@ AdapterProxy* EndpointManager::getConHashProxyForWeight(int64_t hashCode, bool b
         TLOGTARS("[EndpointManager::getConHashProxyForWeight update bStatic:" << bStatic << "|_objName:" << _objName << "|timecost(ms):" << (iEnd - iBegin) << endl);
     }
 
-    if(_consistentHashWeight.size() > 0)
+    while(_consistentHashWeight.size() > 0)
+    {
+        string sNode;
+
+        // 通过一致性hash取到对应的节点
+        _consistentHashWeight.getNodeName(hashCode, sNode);
+
+        auto it = _indexActiveProxys.find(sNode);
+        // 节点不存在，可能是下线或者服务不可用
+        if (it == _indexActiveProxys.end())
+        {
+            updateConHashProxyWeighted(bStatic, _lastConHashWeightProxys, _consistentHashWeight);
+            continue;
+        }
+
+        //被hash到的节点在主控是active的才走在流程
+        if (it->second->isActiveInReg() && it->second->checkActive(true))
+        {
+            return it->second;
+        }
+        else
+        {
+            TLOGWARN("[EndpointManager::getHashProxyForWeight, hash not active," << _objectProxy->name() << "@" << it->second->endpoint().desc() << endl);
+            // 剔除节点再次hash
+            if (!it->second->isActiveInReg())
+            {
+                // 如果在主控的注册状态不是active直接删除，如果状态有变更由updateEndpoints函数里重新添加
+                _indexActiveProxys.erase(sNode);
+            }
+            // checkConHashChange里重新加回到_sortActivProxys重试
+            _sortActivProxys.erase(sNode);
+            updateConHashProxyWeighted(bStatic, _lastConHashWeightProxys, _consistentHashWeight);
+
+            if (_indexActiveProxys.empty())
+            {
+                TLOGERROR("[EndpointManager::getConHashProxyForNormal _activeEndpoints is empty]" << endl);
+                return NULL;
+            }
+        }
+    }
+    /*if(_consistentHashWeight.size() > 0)
     {
         unsigned int iIndex = 0;
 
@@ -1115,6 +1162,7 @@ AdapterProxy* EndpointManager::getConHashProxyForWeight(int64_t hashCode, bool b
         }
         else
         {
+            TLOGWARN("[EndpointManager::getHashProxyForWeight, hash not active," << _objectProxy->name() << "," << _vRegProxys[iIndex]->getTransceiver()->getEndpointInfo().desc() << endl);
             if(_activeProxys.empty())
             {
                 TLOGERROR("[EndpointManager::getConHashProxyForWeight _activeEndpoints is empty], bStatic:" << bStatic << endl);
@@ -1176,7 +1224,7 @@ AdapterProxy* EndpointManager::getConHashProxyForWeight(int64_t hashCode, bool b
 
             return adapterProxy;
         }
-    }
+    }*/
 
     return getHashProxyForNormal(hashCode);
 }
@@ -1203,8 +1251,38 @@ bool EndpointManager::checkHashStaticWeightChange(bool bStatic)
     return false;
 }
 
-bool EndpointManager::checkConHashChange(bool bStatic, const vector<AdapterProxy*> &vLastConHashProxys)
+bool EndpointManager::checkConHashChange(bool bStatic, const map<string, AdapterProxy*> &mLastConHashProxys)
 {
+    // 将之前故障临时剔除的节点重新加回来重试
+    if (_indexActiveProxys.size() != _sortActivProxys.size())
+    {
+        for (auto &it : _indexActiveProxys)
+        {
+            _sortActivProxys[it.first] = it.second;
+        }
+    }
+
+    if(mLastConHashProxys.size() != _sortActivProxys.size())
+    {
+        return true;
+    }
+
+    auto itLast = mLastConHashProxys.begin();
+    auto itSort = _sortActivProxys.begin();
+    for (; itLast!=mLastConHashProxys.end() && itSort!=_sortActivProxys.end(); ++itLast,++itSort)
+    {
+        if (itLast->first != itSort->first)
+        {
+            return true;
+        }
+
+        //解决服务权重更新时一致性哈希环不更新的问题
+        if(bStatic && itSort->second->checkWeightChanged(true))
+        {
+            return true;
+        }
+    }
+/*
     if(vLastConHashProxys.size() != _vRegProxys.size())
     {
         return true;
@@ -1223,6 +1301,7 @@ bool EndpointManager::checkConHashChange(bool bStatic, const vector<AdapterProxy
             return true;
         }
     }
+*/
 
     return false;
 }
@@ -1320,7 +1399,7 @@ void EndpointManager::updateHashProxyWeighted(bool bStatic)
                 _hashStaticRouterCache.push_back(vIndex[i]);
             }
         }
-        
+
         TLOGTARS("EndpointManager::updateHashProxyWeighted bStatic:" << bStatic << "|_objName:" << _objName << "|endpoint:" << vRegProxys[i]->endpoint().desc() << "|iWeight:" << vRegProxys[i]->getWeight() << "|iWeightR:" << iWeight << "|iIndex:" << vIndex[i] << endl);
     }
 
@@ -1354,10 +1433,39 @@ void EndpointManager::updateHashProxyWeighted(bool bStatic)
     }
 }
 
-void EndpointManager::updateConHashProxyWeighted(bool bStatic, vector<AdapterProxy*> &vLastConHashProxys, TC_ConsistentHashNew &conHash)
+void EndpointManager::updateConHashProxyWeighted(bool bStatic, map<string, AdapterProxy*> &mLastConHashProxys, TC_ConsistentHashNew &conHash)
 {
-    if(_vRegProxys.size() <= 0)
+    conHash.clear();
+    if(_sortActivProxys.empty())
     {    
+        TLOGERROR("[EndpointManager::updateHashProxyWeighted _indexActiveProxys is empty], bStatic:" << bStatic << endl);
+        return ;
+    }
+
+    mLastConHashProxys = _sortActivProxys;
+
+    for (auto it = _sortActivProxys.begin(); it != _sortActivProxys.end(); ++it)
+    {
+        int iWeight = (bStatic ? (it->second->getWeight()) : 100);
+        if(iWeight > 0)
+        {
+            iWeight = iWeight / 4;
+            if(iWeight <= 0)
+            {
+                iWeight = 1;
+            }
+            // 同一服务有多个obj的情况
+            // 同一hash值调用不同的obj会hash到不同的服务器
+            // 因为addNode会根据desc(ip+port)计算md5,导致顺序不一致
+            // 一致性hash用host进行索引，不使用index，这里传0
+            conHash.addNode(it->second->endpoint().host(), 0, iWeight);
+        }
+        //防止多个服务节点权重同时更新时一致性哈希环多次更新
+        it->second->resetWeightChanged();
+    }
+/*
+    if(_vRegProxys.size() <= 0)
+    {
         TLOGERROR("[EndpointManager::updateHashProxyWeighted _vRegProxys is empty], bStatic:" << bStatic << endl);
         return ;
     }
@@ -1383,7 +1491,7 @@ void EndpointManager::updateConHashProxyWeighted(bool bStatic, vector<AdapterPro
         //防止多个服务节点权重同时更新时一致性哈希环多次更新
         _vRegProxys[i]->resetWeightChanged();
     }
-
+*/
     conHash.sortNode();
 }
 
@@ -1413,6 +1521,7 @@ AdapterProxy* EndpointManager::getHashProxyForNormal(int64_t hashCode)
     }
     else
     {
+        TLOGWARN("[EndpointManager::getHashProxyForNormal, hash not active," << _objectProxy->name() << "@" << _vRegProxys[hash]->endpoint().desc() << endl);
         if(_activeProxys.empty())
         {
             TLOGERROR("[EndpointManager::getHashProxyForNormal _activeEndpoints is empty]" << endl);
@@ -1494,6 +1603,47 @@ AdapterProxy* EndpointManager::getConHashProxyForNormal(int64_t hashCode)
         TLOGTARS("[EndpointManager::getConHashProxyForNormal update _objName:" << _objName << "|timecost(ms):" << (iEnd - iBegin) << endl);
     }
 
+    while(_consistentHash.size() > 0)
+    {
+        string sNode;
+
+        // 通过一致性hash取到对应的节点
+        _consistentHash.getNodeName(hashCode, sNode);
+
+        auto it = _indexActiveProxys.find(sNode);
+        // 节点不存在，可能是下线或者服务不可用
+        if (it == _indexActiveProxys.end())
+        {
+            updateConHashProxyWeighted(false, _lastConHashProxys, _consistentHash);
+            continue;
+        }
+
+        //被hash到的节点在主控是active的才走在流程
+        if (it->second->isActiveInReg() && it->second->checkActive(true))
+        {
+            return it->second;
+        }
+        else
+        {
+            TLOGWARN("[EndpointManager::getConHashProxyForNormal, hash not active," << _objectProxy->name() << "@" << it->second->endpoint().desc() << endl);
+            // 剔除节点再次hash
+            if (!it->second->isActiveInReg())
+            {
+                // 如果在主控的注册状态不是active直接删除，如果状态有变更由updateEndpoints函数里重新添加
+                _indexActiveProxys.erase(sNode);
+            }
+            // checkConHashChange里重新加回到_sortActivProxys重试
+            _sortActivProxys.erase(sNode);
+            updateConHashProxyWeighted(false, _lastConHashProxys, _consistentHash);
+
+            if (_indexActiveProxys.empty())
+            {
+                TLOGERROR("[EndpointManager::getConHashProxyForNormal _activeEndpoints is empty]" << endl);
+                return NULL;
+            }
+        }
+    }
+/*
     if(_consistentHash.size() > 0)
     {
         unsigned int iIndex = 0;
@@ -1513,6 +1663,7 @@ AdapterProxy* EndpointManager::getConHashProxyForNormal(int64_t hashCode)
         }
         else
         {
+            TLOGWARN("[EndpointManager::getConHashProxyForNormal, hash not active," << _objectProxy->name() << "," << _vRegProxys[iIndex]->getTransceiver()->getEndpointInfo().desc() << endl);
             if(_activeProxys.empty())
             {
                 TLOGERROR("[EndpointManager::getConHashProxyForNormal _activeEndpoints is empty]" << endl);
@@ -1575,7 +1726,7 @@ AdapterProxy* EndpointManager::getConHashProxyForNormal(int64_t hashCode)
             return adapterProxy;
         }
     }
-
+*/
     return getHashProxyForNormal(hashCode);
 }
 
@@ -1998,14 +2149,14 @@ void EndpointManagerThread::getEndpointByAll(vector<EndpointInfo> &activeEndPoin
     pThread->getEndpoints(activeEndPoint,inactiveEndPoint);
 }
 
-void EndpointManagerThread::getEndpointBySet(const string sName, vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint)
+void EndpointManagerThread::getEndpointBySet(const string &sName, vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint)
 {
     EndpointThread * pThread  = getEndpointThread(E_SET,sName);
 
     pThread->getEndpoints(activeEndPoint,inactiveEndPoint);
 }
 
-void EndpointManagerThread::getEndpointByStation(const string sName, vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint)
+void EndpointManagerThread::getEndpointByStation(const string &sName, vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint)
 {
     EndpointThread * pThread  = getEndpointThread(E_STATION,sName);
 
@@ -2027,14 +2178,14 @@ void EndpointManagerThread::getTCEndpointByAll(vector<TC_Endpoint> &activeEndPoi
     pThread->getTCEndpoints(activeEndPoint,inactiveEndPoint);
 }
 
-void EndpointManagerThread::getTCEndpointBySet(const string sName, vector<TC_Endpoint> &activeEndPoint, vector<TC_Endpoint> &inactiveEndPoint)
+void EndpointManagerThread::getTCEndpointBySet(const string &sName, vector<TC_Endpoint> &activeEndPoint, vector<TC_Endpoint> &inactiveEndPoint)
 {
     EndpointThread * pThread  = getEndpointThread(E_SET,sName);
 
     pThread->getTCEndpoints(activeEndPoint,inactiveEndPoint);
 }
 
-void EndpointManagerThread::getTCEndpointByStation(const string sName, vector<TC_Endpoint> &activeEndPoint, vector<TC_Endpoint> &inactiveEndPoint)
+void EndpointManagerThread::getTCEndpointByStation(const string &sName, vector<TC_Endpoint> &activeEndPoint, vector<TC_Endpoint> &inactiveEndPoint)
 {
 
     EndpointThread * pThread  = getEndpointThread(E_STATION,sName);

servant/servant/EndpointManager.h

@@ -418,7 +418,7 @@ private:
     /*
      * 判断静态权重节点是否有变化
      */
-    bool checkConHashChange(bool bStatic, const vector<AdapterProxy*> &vLastConHashProxys);
+    bool checkConHashChange(bool bStatic, const map<string, AdapterProxy*> &mLastConHashProxys);
 
     /*
      * 更新取模hash方法的静态权重节点信息
@@ -428,7 +428,7 @@ private:
     /*
      * 更新一致性hash方法的静态权重节点信息
      */
-    void updateConHashProxyWeighted(bool bStatic, vector<AdapterProxy*> &vLastConHashProxys, TC_ConsistentHashNew &conHash);
+    void updateConHashProxyWeighted(bool bStatic, map<string, AdapterProxy*> &mLastConHashProxys, TC_ConsistentHashNew &conHash);
 
     /*
      * 根据后端服务的权重值选取一个结点
@@ -466,7 +466,14 @@ private:
      * 活跃的结点
      */
     vector<AdapterProxy*>         _activeProxys;
-    
+
+    /*
+     * 一致性hash使用，保证强一致性
+     * key：host
+     */
+    map<string, AdapterProxy*>            _sortActivProxys;
+    unordered_map<string, AdapterProxy*>  _indexActiveProxys;
+
     /*
      * 部署的结点 包括活跃的和不活跃的
      */
@@ -532,7 +539,7 @@ private:
     /*
      * 一致性hash静态权重时使用
      */
-    vector<AdapterProxy*>         _lastConHashWeightProxys;
+    map<string, AdapterProxy*>    _lastConHashWeightProxys;
 
     /*
      * 一致性hash静态权重时使用
@@ -542,7 +549,7 @@ private:
     /*
      * 一致性hash普通使用
      */
-    vector<AdapterProxy*>         _lastConHashProxys;
+    map<string, AdapterProxy*>    _lastConHashProxys;
 
     /*
      * 一致性hash普通使用
@@ -666,12 +673,12 @@ public:
     /*
      * 根据set获取可用与不可用的结点
      */
-    void getEndpointBySet(const string sName, vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint);
+    void getEndpointBySet(const string &sName, vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint);
 
     /*
      * 根据地区获取可用与不可用的结点
      */
-    void getEndpointByStation(const string sName, vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint);
+    void getEndpointByStation(const string &sName, vector<EndpointInfo> &activeEndPoint, vector<EndpointInfo> &inactiveEndPoint);
 
     /*
      * 按idc获取可用与不可用的结点
@@ -686,12 +693,12 @@ public:
     /*
      * 根据set获取可用与不可用的结点
      */
-    void getTCEndpointBySet(const string sName, vector<TC_Endpoint> &activeEndPoint, vector<TC_Endpoint> &inactiveEndPoint);
+    void getTCEndpointBySet(const string &sName, vector<TC_Endpoint> &activeEndPoint, vector<TC_Endpoint> &inactiveEndPoint);
 
     /*
      * 根据地区获取可用与不可用的结点
      */
-    void getTCEndpointByStation(const string sName, vector<TC_Endpoint> &activeEndPoint, vector<TC_Endpoint> &inactiveEndPoint);
+    void getTCEndpointByStation(const string &sName, vector<TC_Endpoint> &activeEndPoint, vector<TC_Endpoint> &inactiveEndPoint);
 
 protected:
 

unit-test/hello_test.cpp

@@ -707,7 +707,7 @@ void HelloTest::rpcFromRegistry(Communicator *comm)
 
 	//test rpc loop
 	int count = 10;
-	while(count-- > 10)
+	while(count-- > 0)
 	{
 		int ret;
 		string out1, out2;
@@ -721,7 +721,7 @@ void HelloTest::rpcFromRegistry(Communicator *comm)
 
 	//test rpc hash
 	count = 10;
-	while(count-- > 10)
+	while(count-- > 0)
 	{
 		vector<string> vout;
 
@@ -742,6 +742,31 @@ void HelloTest::rpcFromRegistry(Communicator *comm)
 
 }
 
+void HelloTest::rpcConHashFromRegistry(Communicator *comm)
+{
+	HelloPrx prx = comm->stringToProxy<HelloPrx>("TestApp.RpcServer.HelloObj");
+
+	//test rpc consistent hash
+	int count = 10;
+	while(count-- > 0)
+	{
+		std::thread conHash([&](){
+			std::set<std::string> servInfos;
+			for (int j = 0; j < 10; ++j) {
+				std::string serverInfo;
+				int ret = prx->tars_consistent_hash(TC_Thread::CURRENT_THREADID())->testConHash(serverInfo);
+				LOG_CONSOLE_DEBUG << "hashCode:" << TC_Thread::CURRENT_THREADID() << ", serverInfo:" << serverInfo << endl;
+				servInfos.emplace(serverInfo);
+				ASSERT_TRUE(ret == 0);
+			}
+
+			ASSERT_TRUE(servInfos.size() == 1);
+     		});
+
+		conHash.join();
+	}
+}
+
 void HelloTest::checkSyncTimeout(Communicator *comm)
 {
 	HelloPrx prx = getObj<HelloPrx>(comm, "HelloAdapter");

unit-test/hello_test.h

@@ -328,6 +328,8 @@ public:
 
 	void rpcFromRegistry(Communicator *comm);
 
+	void rpcConHashFromRegistry(Communicator *comm);
+
 	void forEach(function<void()> func);
 
 	void forEach(function<void(Communicator *comm)> func);

unit-test/rpc/test_registry.cpp

@@ -195,6 +195,22 @@ TEST_F(HelloTest, registryRpc)
 }
 
 
+//一致hash测试
+TEST_F(HelloTest, registryRpcConHashInvoke)
+{
+	START_FRAMEWORK_SERVER_1_2
+        CDbHandle::addInactiveEndPoint("TestApp.RpcServer.HelloObj", 9989, 1);
+	CDbHandle::addInactiveEndPoint("TestApp.RpcServer.HelloObj", 9992, 1);
+
+	shared_ptr<Communicator> c = getCommunicator();
+
+	rpcConHashFromRegistry(c.get());
+
+	stopServer(rpc1Server);
+	stopServer(rpc2Server);
+        STOP_FRAMEWORK_SERVER;
+}
+
 TEST_F(HelloTest, registryRpcUpdateList)
 {
 	shared_ptr<Communicator> comm = getCommunicator();

unit-test/server/Hello.tars

@@ -60,6 +60,7 @@ module Test
 
         bool testCoro(string sIn, out string sOut);
 
+        int testConHash(out string sOut);
     };
 };
 

unit-test/server/HelloImp.cpp

@@ -147,4 +147,10 @@ bool HelloImp::testCoro(const std::string& sIn, std::string &sOut, CurrentPtr cu
 	return ServantProxyThreadData::getData()->_sched != NULL;
 }
 
+int HelloImp::testConHash(std::string &sOut, CurrentPtr current)
+{
+	sOut = _handle->getBindAdapter()->getEndpoint().toString();
+	return 0;
+}
+
 ///////////////////////////////////////////////////////////////////////////////

unit-test/server/HelloImp.h

@@ -47,6 +47,8 @@ public:
 
 	virtual bool testCoro(const std::string& sIn, std::string &sOut, CurrentPtr current);
 
+        int testConHash(std::string &sOut, CurrentPtr current);
+
 protected:
     HelloPrx _helloPrx;
 

util/include/util/tc_consistent_hash_new.h

@@ -96,6 +96,12 @@ public:
         */
         int32_t iHashCode;
 
+        /**
+         *节点名称
+         *node name
+         */
+        string sNode;
+
         /**
          *节点下标
          * node subscript
@@ -161,6 +167,19 @@ public:
      */
     int getIndex(const string & key, unsigned int & iIndex);
 
+    /**
+     * @brief 获取某key对应到的节点node的名称.
+     * @brief Gets the name of the node to which a key corresponds.
+     *
+     * @param key      key名称
+     * @param key      key name
+     * @param sNode    对应到的节点的名称
+     * @param sNode    the name of the node to which corresponds.
+     * @return        0:获取成功   -1:没有被添加的节点
+     * @return        0:obtain successfully  -1:no nodes added
+     */
+    int getNodeName(const string & key, string & sNode);
+
     /**
      * @brief 获取某hashcode对应到的节点node的下标.
      * @brief Gets the subscript of the node to which a certain hashcode corresponds
@@ -173,6 +192,18 @@ public:
      */
     int getIndex(int32_t hashcode, unsigned int & iIndex);
 
+    /**
+     * @brief 获取某hashcode对应到的节点node的名称.
+     * @brief Gets the name of the node to which a certain hashcode corresponds
+     *
+     * @param hashcode      hashcode
+     * @param sNode   对应到的节点的名称
+     * @param sNode   the name of the node to which corresponds.
+     * @return        0:获取成功   -1:没有被添加的节点
+     * @return        0:obtain successfully  -1:no nodes added
+     */
+    int getNodeName(int32_t hashcode, string & sNode);
+
     /**
      * @brief 获取当前hash列表的长度.
      * @brief Get the length of the current hash list

util/src/tc_consistent_hash_new.cpp

@@ -126,7 +126,7 @@ void TC_ConsistentHashNew::printNode()
             mapNode[_vHashList[i].iIndex] = value;
         }
 
-        cout << "printNode: " << _vHashList[i].iHashCode << "|" << _vHashList[i].iIndex << "|" << mapNode[_vHashList[i].iIndex] << endl;
+        cout << "printNode: " << _vHashList[i].iHashCode << "|" << _vHashList[i].sNode << "|" << _vHashList[i].iIndex << "|" << mapNode[_vHashList[i].iIndex] << endl;
     }
 
     map<unsigned int, unsigned int>::iterator it = mapNode.begin();
@@ -152,6 +152,7 @@ int TC_ConsistentHashNew::addNode(const string & node, unsigned int index, int w
     }
 
     node_T_new stItem;
+    stItem.sNode = node;
     stItem.iIndex = index;
 
     for (int j = 0; j < weight; j++)
@@ -237,4 +238,55 @@ int TC_ConsistentHashNew::getIndex(int32_t hashcode, unsigned int & iIndex)
     return 0;
 }
 
+int TC_ConsistentHashNew::getNodeName(const string & key, string & sNode)
+{
+    if(_ptrHashAlg.get() == NULL || _vHashList.size() == 0)
+    {
+        sNode = "";
+        return -1;
+    }
+
+    vector<char> data = TC_MD5::md5bin(key);
+    int32_t iCode = _ptrHashAlg->hash(data.data(), data.size());
+
+    return getNodeName(iCode, sNode);
+}
+
+int TC_ConsistentHashNew::getNodeName(int32_t hashcode, string & sNode)
+{
+    if(_ptrHashAlg.get() == NULL || _vHashList.size() == 0)
+    {
+        sNode = "";
+        return -1;
+    }
+
+    // 只保留32位
+    int32_t iCode = (hashcode & 0xFFFFFFFFL);
+
+    int low = 0;
+    int high = (int)_vHashList.size();
+
+    if(iCode <= _vHashList[0].iHashCode || iCode > _vHashList[high-1].iHashCode)
+    {
+        sNode = _vHashList[0].sNode;
+        return 0;
+    }
+
+
+    while (low < high - 1)
+    {
+        int mid = (low + high) / 2;
+        if (_vHashList[mid].iHashCode > iCode)
+        {
+            high = mid;
+        }
+        else
+        {
+            low = mid;
+        }
+    }
+    sNode = _vHashList[low+1].sNode;
+    return 0;
+}
+
 }