尧图网站建设 
引言本篇文章基于分布式存储的背景一致性哈希是分布式存储里面的一种技术用于多个结点之间分布数据以最大程度减少添加或删除结点时候的数据重组我们在使用普通的哈希的时候使用hash(key) % n但是一旦n这个结点数量发生变化的时候我们所有的数据位置都会发生变化所以会导致所有的数据都会被移动数据少还可以接受但是对于分布式存储来说我们需要存储很多的数据这样子效率会非常低。而一致性哈希就解决了这样一个问题可以保证当结点数量发生变化的时候只需要更改一小部分映射键就可以解决。一致性哈希的结构一致性哈希就是一个环尾部连接着头部环上的所有结点全部都是虚拟结点而真实的结点被我们存储在另一个地方我们可以假设有A、B、C三个真实结点而环上的我们可以用A-1A-2B-1B-2C-1C-2实际上这些结点全部都是int类型方便hash来定位数据来表示。这些虚拟结点全部都映射着唯一一个真实的结点而我们的数据是通过虚拟结点来找到真实结点来存储。我们为什么需要虚拟结点呢均匀分布数据因为我们的key是通过hash来映射分布的所以为了让数据分布的比较均匀虚拟结点可以避免负载不均衡。减少数据的迁移当结点增减的时候因为每个虚拟结点只需要负责一小部分所以结点发生变化的时候只需要将虚拟结点附近的数据重新分配减少了整体的迁移量。访问这些虚拟结点实际上还是访问的真实结点我们数据怎么存储在这个环上呢其实就是每一个结点都有一个key值string类型数据通过我们自定义的Crc32IEEETable来找到其对应的虚拟结点这个结构一旦有一个结点失效了要删除的时候这个真实结点对应的所有虚拟结点删除之后因为是环状的结构所以那些数据就会对应到顺着的下一个结点而其他那些数据根本不受影响所有的数据并不是直接连在一个结点上的而是绑定到了下一个结点比如keys_ [10, 30, 60, 90] hash 30 → 找到 30 hash 45 → 找到 60没有相等的值也能找到 hash 95 → 返回 end()绕回 10这样子当30消失了那么只需要把和30相关的数据全部转移到60上面就可以这主要的原因是因为我们找数据的方式是二分查找这样子我们得到的指针一定是指向下一个结点代码每一个一致性哈希都需要一些配置信息比如最大的虚拟结点个数最小的虚拟结点个数这里的hash_func是一个可配置的哈希函数用来将字符串映射成uint32_t类型的哈希值在一致性哈希中用于确定节点或 key 在哈希环上的位置// 一致性哈希的配置 struct HashConfig { int replicas; // 每个真实结点对应的虚拟结点个数 int min_replicas; int max_replicas; std::functionuint32_t(const std::string_view) hash_func; double load_balance_threadshold; // 负载均衡值超过此值触发虚拟节点调整 };这个就是我们的默认配置const HashConfig DefaultConfig {10, 10, 200, [](const std::string_view data) - uint32_t {return Crc32IEEE(data);}, 0.25};我们的一致性哈希主要就是增加删除获取结点得到一致性哈希的负载信息。其哈希环上的结点都是int32的类型这样可以由hash来找到确定的虚拟结点然后每一个虚拟结点都必须要对应一个真实结点这个我们通过map来一一映射真实结点是string类型因为真实结点的名字有着特殊的意义除了虚拟结点对于真实结点的映射还有真实结点对于虚拟结点的存储每一个真实结点都需要存储自己所有的虚拟结点方便之后的负载均衡最后就是要记录一些访问节点的信息用来做出最及时的调整。// 虚拟结点用数字表示真实结点用字符表示 class ConsistentHashMap { public: explicit ConsistentHashMap(HashConfig cfg DefaultConfig); ~ConsistentHashMap(); [[nodiscard]] bool Add(const std::vectorstd::string_view nodes); [[nodiscard]] bool Remove(const std::string_view node); // 获取结点 [[nodiscard]] auto Get(const std::string_view key) - std::string_view; // 获取负载统计信息 [[nodiscard]] auto GetStarts() - std::unordered_mapstd::string, double; private: // 添加结点的虚拟结点 void AddNode(const std::string_view node, int replicas); // 检查并且重新平衡虚拟结点 void CheckAndRebalance(); // 重新平衡结点 void RebalanceNodes(); // 启动负载均衡器线程 void StartBalancer(); mutable std::shared_mutex mutex_; // mutable让 mutex_ 可以在 const 成员函数里被修改 HashConfig config_; std::vectoruint32_t keys_; // 哈希环没有真实结点 std::unordered_mapuint32_t, std::string_view hash_map_; // 哈希环虚拟结点到真实结点的映射 std::unordered_mapstd::string_view, int node_replicas_; // 每个真实节点对应多少个虚拟节点 std::unordered_mapstd::string_view, std::atomiclong long node_counts_; // 节点被选中了多少次即请求计数 std::atomiclong long total_requests_; // 总请求数 std::thread balancer_thread_; // 负载均衡器的线程 std::atomicbool is_balancer_stop_; // 负载均衡器线程停止的标志 };因为我们的数据都是string类型的所以通过string类型找到对应的hash_key有一个Crc表static constexpr uint32_t Crc32IEEETable[256] { 0x00000000, 0x77073096, 0xee0e612c, 0x990951ba, 0x076dc419, 0x706af48f, 0xe963a535, 0x9e6495a3, 0x0edb8832, 0x79dcb8a4, 0xe0d5e91e, 0x97d2d988, 0x09b64c2b, 0x7eb17cbd, 0xe7b82d07, 0x90bf1d91, 0x1db71064, 0x6ab020f2, 0xf3b97148, 0x84be41de, 0x1adad47d, 0x6ddde4eb, 0xf4d4b551, 0x83d385c7, 0x136c9856, 0x646ba8c0, 0xfd62f97a, 0x8a65c9ec, 0x14015c4f, 0x63066cd9, 0xfa0f3d63, 0x8d080df5, 0x3b6e20c8, 0x4c69105e, 0xd56041e4, 0xa2677172, 0x3c03e4d1, 0x4b04d447, 0xd20d85fd, 0xa50ab56b, 0x35b5a8fa, 0x42b2986c, 0xdbbbc9d6, 0xacbcf940, 0x32d86ce3, 0x45df5c75, 0xdcd60dcf, 0xabd13d59, 0x26d930ac, 0x51de003a, 0xc8d75180, 0xbfd06116, 0x21b4f4b5, 0x56b3c423, 0xcfba9599, 0xb8bda50f, 0x2802b89e, 0x5f058808, 0xc60cd9b2, 0xb10be924, 0x2f6f7c87, 0x58684c11, 0xc1611dab, 0xb6662d3d, 0x76dc4190, 0x01db7106, 0x98d220bc, 0xefd5102a, 0x71b18589, 0x06b6b51f, 0x9fbfe4a5, 0xe8b8d433, 0x7807c9a2, 0x0f00f934, 0x9609a88e, 0xe10e9818, 0x7f6a0dbb, 0x086d3d2d, 0x91646c97, 0xe6635c01, 0x6b6b51f4, 0x1c6c6162, 0x856530d8, 0xf262004e, 0x6c0695ed, 0x1b01a57b, 0x8208f4c1, 0xf50fc457, 0x65b0d9c6, 0x12b7e950, 0x8bbeb8ea, 0xfcb9887c, 0x62dd1ddf, 0x15da2d49, 0x8cd37cf3, 0xfbd44c65, 0x4db26158, 0x3ab551ce, 0xa3bc0074, 0xd4bb30e2, 0x4adfa541, 0x3dd895d7, 0xa4d1c46d, 0xd3d6f4fb, 0x4369e96a, 0x346ed9fc, 0xad678846, 0xda60b8d0, 0x44042d73, 0x33031de5, 0xaa0a4c5f, 0xdd0d7cc9, 0x5005713c, 0x270241aa, 0xbe0b1010, 0xc90c2086, 0x5768b525, 0x206f85b3, 0xb966d409, 0xce61e49f, 0x5edef90e, 0x29d9c998, 0xb0d09822, 0xc7d7a8b4, 0x59b33d17, 0x2eb40d81, 0xb7bd5c3b, 0xc0ba6cad, 0xedb88320, 0x9abfb3b6, 0x03b6e20c, 0x74b1d29a, 0xead54739, 0x9dd277af, 0x04db2615, 0x73dc1683, 0xe3630b12, 0x94643b84, 0x0d6d6a3e, 0x7a6a5aa8, 0xe40ecf0b, 0x9309ff9d, 0x0a00ae27, 0x7d079eb1, 0xf00f9344, 0x8708a3d2, 0x1e01f268, 0x6906c2fe, 0xf762575d, 0x806567cb, 0x196c3671, 0x6e6b06e7, 0xfed41b76, 0x89d32be0, 0x10da7a5a, 0x67dd4acc, 0xf9b9df6f, 0x8ebeeff9, 0x17b7be43, 0x60b08ed5, 0xd6d6a3e8, 0xa1d1937e, 0x38d8c2c4, 0x4fdff252, 0xd1bb67f1, 0xa6bc5767, 0x3fb506dd, 0x48b2364b, 0xd80d2bda, 0xaf0a1b4c, 0x36034af6, 0x41047a60, 0xdf60efc3, 0xa867df55, 0x316e8eef, 0x4669be79, 0xcb61b38c, 0xbc66831a, 0x256fd2a0, 0x5268e236, 0xcc0c7795, 0xbb0b4703, 0x220216b9, 0x5505262f, 0xc5ba3bbe, 0xb2bd0b28, 0x2bb45a92, 0x5cb36a04, 0xc2d7ffa7, 0xb5d0cf31, 0x2cd99e8b, 0x5bdeae1d, 0x9b64c2b0, 0xec63f226, 0x756aa39c, 0x026d930a, 0x9c0906a9, 0xeb0e363f, 0x72076785, 0x05005713, 0x95bf4a82, 0xe2b87a14, 0x7bb12bae, 0x0cb61b38, 0x92d28e9b, 0xe5d5be0d, 0x7cdcefb7, 0x0bdbdf21, 0x86d3d2d4, 0xf1d4e242, 0x68ddb3f8, 0x1fda836e, 0x81be16cd, 0xf6b9265b, 0x6fb077e1, 0x18b74777, 0x88085ae6, 0xff0f6a70, 0x66063bca, 0x11010b5c, 0x8f659eff, 0xf862ae69, 0x616bffd3, 0x166ccf45, 0xa00ae278, 0xd70dd2ee, 0x4e048354, 0x3903b3c2, 0xa7672661, 0xd06016f7, 0x4969474d, 0x3e6e77db, 0xaed16a4a, 0xd9d65adc, 0x40df0b66, 0x37d83bf0, 0xa9bcae53, 0xdebb9ec5, 0x47b2cf7f, 0x30b5ffe9, 0xbdbdf21c, 0xcabac28a, 0x53b39330, 0x24b4a3a6, 0xbad03605, 0xcdd70693, 0x54de5729, 0x23d967bf, 0xb3667a2e, 0xc4614ab8, 0x5d681b02, 0x2a6f2b94, 0xb40bbe37, 0xc30c8ea1, 0x5a05df1b, 0x2d02ef8d, }; uint32_t Crc32IEEE(const std::string data) { uint32_t crc 0xFFFFFFFF; for (char c : data) { crc Crc32IEEETable[(crc ^ static_castuint8_t(c)) 0xFF] ^ (crc 8); } return crc ^ 0xFFFFFFFF; }添加节点调用的其实就是AddNode添加结点是需要写入操作的所以我们用的是写锁然后遍历所有我们需要添加的结点并且同时我们需要把配置信息里面的当前虚拟结点数量传入进去因为新添加一个真实结点就需要添加对应的虚拟节点而新添加的虚拟节点我们会先对这个结点起一个有规律的名字我们通过自定义的Crc转化为int类型放入vector容器里面可能最后的容器里面是1 3 2 4 6 5 这样子没有顺序的样子所以我们需要重排而重排的目的就是为了之后得到数据的时候可以通过二分查找来快速找到key值bool ConsistentHashMap::Add(const std::vectorstd::string_view nodes) { if (nodes.empty()) { return false; } std::unique_lock lock{mutex_}; // 写锁 for (const auto node : nodes) { if (node.empty()) { continue; } AddNode(node, config_.replicas); } // 重排哈希环 std::sort(keys_.begin(), keys_.end()); return true; } void ConsistentHashMap::AddNode(const std::string_view node, int replicas) { for (int i 0; i replicas; i) { std::string hash_key fmt::format({}-{}, node, std::to_string(i)); uint32_t hash config_.hash_func(hash_key); keys_.push_back(hash); hash_map_[hash] node; } node_replicas_[node] replicas; // 如果这个结点是新添加的初始化其计数器 if (node_counts_.find(node) node_counts_.end()) { node_counts_[node] 0; } }移除结点就是要先删除虚拟结点然后删除对应的真实结点删除虚拟结点的办法就是根据我们之前的规律找到对应的key然后将其删除。我们这里使用的是remove erase的操作首先remove并不是删除其作用就是移动元素当执行完remove之后会返回一个指向新数组尾部的一个指针我们就可以利用这个指针来删除原来旧容器里面那些没有用的数据所以我们需要remove erase最后我们需要删除真实结点nodebool ConsistentHashMap::Remove(const std::string_view node) // 传入的就是真实结点 { if (node.empty()) { return false; } std::unique_lock lock{mutex_}; auto it_replicas node_replicas_.find(node); // 找虚拟映射结点 if (it_replicas node_replicas_.end()) { return false; } int replicas it_replicas-second; // 移除结点的所有虚拟节点 for (int i 0; i replicas; i) { std::string hash_key fmt::format({}-{}, node.data(), std::to_string(i)); uint32_t hash config_.hash_func(hash_key); hash_map_.erase(hash); auto it std::remove(keys_.begin(), keys_.end(), hash); keys_.erase(it, keys_.end()); } // 删除真实结点 node_replicas_.erase(node); node_counts_.erase(node); return true; }我们要查询数据的时候用二分查找这也是我们为什么之前需要对keys进行排序并且一定要记住我们一致性哈希是一个环所以当我们找到末尾的时候其实就是开头。然后我们通过虚拟结点找到对应的真实结点auto ConsistentHashMap::Get(const std::string_view key) - std::string_view { if (key.empty()) { return ; } if (keys_.empty()) { return ; } uint32_t hash config_.hash_func(key); // 二分查找 auto it std::lower_bound(keys_.begin(), keys_.end(), hash); // 处理边界情况 : 如果到了末尾则回到开头 if (it keys_.end()) { it keys_.begin(); } std::string_view node hash_map_[*it]; // 增加结点计数和总数请求数 node_counts_[node]; total_requests_; return node; }统计负载信息然后返回一个数组这个数组里面存储的是每一个结点的负载信息auto ConsistentHashMap::GetStarts() - std::unordered_mapstd::string, double { std::shared_lock lock{mutex_}; std::unordered_mapstd::string, double stats; long long curr_total total_requests_.load(); if (curr_total 0) { return stats; } for (auto const [node, count] : node_counts_) { stats[node.data()] static_castdouble(count.load()) / static_castdouble(curr_total); } return stats; }开启负载均衡其实就是创建一个新的线程这个线程里面专门执行负载均衡的检查和修正首先是这个检查其实就是计算当前结点里面最高的差异diff如果这个diff比我们设置的threadshold要大那么我们就应该修正了void ConsistentHashMap::StartBalancer() { is_balancer_stop_ false; balancer_thread_ std::thread{ [this] { while(!is_balancer_stop_) { std::this_thread::sleep_for(std::chrono::seconds(1)); if (!is_balancer_stop_) // 再次检查防止在sleep期间被要求停止 { CheckAndRebalance(); } } }}; } void ConsistentHashMap::CheckAndRebalance() { if (total_requests_.load() 1000) { return; // 样本太少了不进行调整 } std::shared_lock lock{mutex_}; if (node_replicas_.empty()) { return; } // 计算系统平均负载总请求数 / 物理节点数 long long current_total_requests total_requests_.load(); // 每台真实服务器的平均请求数用来判断哪台服务器负载过高 double avg_load static_castdouble(current_total_requests) / node_replicas_.size(); double max_diff 0.0; for (auto const [node, count] : node_counts_) { double diff std::abs(static_castdouble(count.load()) - avg_load); if (avg_load 0) { if (diff / avg_load max_diff) { max_diff diff / avg_load; } } else if (diff 0) // 如果avg是0但是counts不是0说明不平衡 { max_diff 1.0; } } lock.unlock(); // 释放读锁因为rebalenceNodes要写锁 if (max_diff config_.load_balance_threadshold) { RebalanceNodes(); } }这个就是平衡的措施这个平衡措施不是针对某一个结点而是所有的结点移除全部的结点之后再重新添加结点但是因为AddNode里面没有排序的措施所以我们还需要补充sort对于keys的排序void ConsistentHashMap::RebalanceNodes() { std::unique_lock lock{mutex_}; if (node_replicas_.empty()) { return; } long long current_total_requests total_requests_.load(); double avg_load static_castdouble(current_total_requests) / node_replicas_.size(); // 调整每个结点的虚拟结点数量 // 注意这里需要创建一个副本因为在循环里面可能会修改 nodeReplicas 和 nodeCounts std::unordered_mapstd::string_view, int curr_replicas node_replicas_; std::unordered_mapstd::string_view, std::atomiclong long curr_counts; for (auto const [node, count] : node_counts_) { curr_counts[node] count.load(); } for (auto const [node, count] : curr_counts) { int old_replicas curr_replicas[node]; double load_ratio 0.0; if (avg_load 0) { load_ratio static_castdouble(count) / avg_load; } else if (count 0) { load_ratio 2.0; } else { load_ratio 1.0; } int new_replicas; if (load_ratio 1.0) { // 负载过高减少虚拟结点 new_replicas static_castint(std::round(static_castdouble(old_replicas) / load_ratio)); } else { new_replicas static_castint(std::round(static_castdouble(old_replicas) * (2.0 - load_ratio))); } // 确保在有限的范围之内 if (new_replicas config_.min_replicas) { new_replicas config_.min_replicas; } else if (new_replicas config_.max_replicas) { new_replicas config_.max_replicas; } if (new_replicas ! old_replicas) { // 重新添加结点的虚拟节点先移除旧的结点再添加新的 // 移除结点的所有虚拟结点 int replicas_to_remove node_replicas_[node]; for (int i 0; i replicas_to_remove; i) { std::string hashKey static_caststd::string(node) - std::to_string(i); uint32_t hash config_.hash_func(hashKey); hash_map_.erase(hash); auto it std::remove(keys_.begin(), keys_.end(), hash); keys_.erase(it, keys_.end()); } node_replicas_.erase(node); AddNode(node, new_replicas); } } // 重置计数器 for (auto pair : node_counts_) { pair.second.store(0); } total_requests_.store(0); std::sort(keys_.begin(), keys_.end()); }本篇文章到这里就结束了希望可以帮助大家理解~~~ 一致性哈希
引言本篇文章基于分布式存储的背景一致性哈希是分布式存储里面的一种技术用于多个结点之间分布数据以最大程度减少添加或删除结点时候的数据重组我们在使用普通的哈希的时候使用hash(key) % n但是一旦n这个结点数量发生变化的时候我们所有的数据位置都会发生变化所以会导致所有的数据都会被移动数据少还可以接受但是对于分布式存储来说我们需要存储很多的数据这样子效率会非常低。而一致性哈希就解决了这样一个问题可以保证当结点数量发生变化的时候只需要更改一小部分映射键就可以解决。一致性哈希的结构一致性哈希就是一个环尾部连接着头部环上的所有结点全部都是虚拟结点而真实的结点被我们存储在另一个地方我们可以假设有A、B、C三个真实结点而环上的我们可以用A-1A-2B-1B-2C-1C-2实际上这些结点全部都是int类型方便hash来定位数据来表示。这些虚拟结点全部都映射着唯一一个真实的结点而我们的数据是通过虚拟结点来找到真实结点来存储。我们为什么需要虚拟结点呢均匀分布数据因为我们的key是通过hash来映射分布的所以为了让数据分布的比较均匀虚拟结点可以避免负载不均衡。减少数据的迁移当结点增减的时候因为每个虚拟结点只需要负责一小部分所以结点发生变化的时候只需要将虚拟结点附近的数据重新分配减少了整体的迁移量。访问这些虚拟结点实际上还是访问的真实结点我们数据怎么存储在这个环上呢其实就是每一个结点都有一个key值string类型数据通过我们自定义的Crc32IEEETable来找到其对应的虚拟结点这个结构一旦有一个结点失效了要删除的时候这个真实结点对应的所有虚拟结点删除之后因为是环状的结构所以那些数据就会对应到顺着的下一个结点而其他那些数据根本不受影响所有的数据并不是直接连在一个结点上的而是绑定到了下一个结点比如keys_ [10, 30, 60, 90] hash 30 → 找到 30 hash 45 → 找到 60没有相等的值也能找到 hash 95 → 返回 end()绕回 10这样子当30消失了那么只需要把和30相关的数据全部转移到60上面就可以这主要的原因是因为我们找数据的方式是二分查找这样子我们得到的指针一定是指向下一个结点代码每一个一致性哈希都需要一些配置信息比如最大的虚拟结点个数最小的虚拟结点个数这里的hash_func是一个可配置的哈希函数用来将字符串映射成uint32_t类型的哈希值在一致性哈希中用于确定节点或 key 在哈希环上的位置// 一致性哈希的配置 struct HashConfig { int replicas; // 每个真实结点对应的虚拟结点个数 int min_replicas; int max_replicas; std::functionuint32_t(const std::string_view) hash_func; double load_balance_threadshold; // 负载均衡值超过此值触发虚拟节点调整 };这个就是我们的默认配置const HashConfig DefaultConfig {10, 10, 200, [](const std::string_view data) - uint32_t {return Crc32IEEE(data);}, 0.25};我们的一致性哈希主要就是增加删除获取结点得到一致性哈希的负载信息。其哈希环上的结点都是int32的类型这样可以由hash来找到确定的虚拟结点然后每一个虚拟结点都必须要对应一个真实结点这个我们通过map来一一映射真实结点是string类型因为真实结点的名字有着特殊的意义除了虚拟结点对于真实结点的映射还有真实结点对于虚拟结点的存储每一个真实结点都需要存储自己所有的虚拟结点方便之后的负载均衡最后就是要记录一些访问节点的信息用来做出最及时的调整。// 虚拟结点用数字表示真实结点用字符表示 class ConsistentHashMap { public: explicit ConsistentHashMap(HashConfig cfg DefaultConfig); ~ConsistentHashMap(); [[nodiscard]] bool Add(const std::vectorstd::string_view nodes); [[nodiscard]] bool Remove(const std::string_view node); // 获取结点 [[nodiscard]] auto Get(const std::string_view key) - std::string_view; // 获取负载统计信息 [[nodiscard]] auto GetStarts() - std::unordered_mapstd::string, double; private: // 添加结点的虚拟结点 void AddNode(const std::string_view node, int replicas); // 检查并且重新平衡虚拟结点 void CheckAndRebalance(); // 重新平衡结点 void RebalanceNodes(); // 启动负载均衡器线程 void StartBalancer(); mutable std::shared_mutex mutex_; // mutable让 mutex_ 可以在 const 成员函数里被修改 HashConfig config_; std::vectoruint32_t keys_; // 哈希环没有真实结点 std::unordered_mapuint32_t, std::string_view hash_map_; // 哈希环虚拟结点到真实结点的映射 std::unordered_mapstd::string_view, int node_replicas_; // 每个真实节点对应多少个虚拟节点 std::unordered_mapstd::string_view, std::atomiclong long node_counts_; // 节点被选中了多少次即请求计数 std::atomiclong long total_requests_; // 总请求数 std::thread balancer_thread_; // 负载均衡器的线程 std::atomicbool is_balancer_stop_; // 负载均衡器线程停止的标志 };因为我们的数据都是string类型的所以通过string类型找到对应的hash_key有一个Crc表static constexpr uint32_t Crc32IEEETable[256] { 0x00000000, 0x77073096, 0xee0e612c, 0x990951ba, 0x076dc419, 0x706af48f, 0xe963a535, 0x9e6495a3, 0x0edb8832, 0x79dcb8a4, 0xe0d5e91e, 0x97d2d988, 0x09b64c2b, 0x7eb17cbd, 0xe7b82d07, 0x90bf1d91, 0x1db71064, 0x6ab020f2, 0xf3b97148, 0x84be41de, 0x1adad47d, 0x6ddde4eb, 0xf4d4b551, 0x83d385c7, 0x136c9856, 0x646ba8c0, 0xfd62f97a, 0x8a65c9ec, 0x14015c4f, 0x63066cd9, 0xfa0f3d63, 0x8d080df5, 0x3b6e20c8, 0x4c69105e, 0xd56041e4, 0xa2677172, 0x3c03e4d1, 0x4b04d447, 0xd20d85fd, 0xa50ab56b, 0x35b5a8fa, 0x42b2986c, 0xdbbbc9d6, 0xacbcf940, 0x32d86ce3, 0x45df5c75, 0xdcd60dcf, 0xabd13d59, 0x26d930ac, 0x51de003a, 0xc8d75180, 0xbfd06116, 0x21b4f4b5, 0x56b3c423, 0xcfba9599, 0xb8bda50f, 0x2802b89e, 0x5f058808, 0xc60cd9b2, 0xb10be924, 0x2f6f7c87, 0x58684c11, 0xc1611dab, 0xb6662d3d, 0x76dc4190, 0x01db7106, 0x98d220bc, 0xefd5102a, 0x71b18589, 0x06b6b51f, 0x9fbfe4a5, 0xe8b8d433, 0x7807c9a2, 0x0f00f934, 0x9609a88e, 0xe10e9818, 0x7f6a0dbb, 0x086d3d2d, 0x91646c97, 0xe6635c01, 0x6b6b51f4, 0x1c6c6162, 0x856530d8, 0xf262004e, 0x6c0695ed, 0x1b01a57b, 0x8208f4c1, 0xf50fc457, 0x65b0d9c6, 0x12b7e950, 0x8bbeb8ea, 0xfcb9887c, 0x62dd1ddf, 0x15da2d49, 0x8cd37cf3, 0xfbd44c65, 0x4db26158, 0x3ab551ce, 0xa3bc0074, 0xd4bb30e2, 0x4adfa541, 0x3dd895d7, 0xa4d1c46d, 0xd3d6f4fb, 0x4369e96a, 0x346ed9fc, 0xad678846, 0xda60b8d0, 0x44042d73, 0x33031de5, 0xaa0a4c5f, 0xdd0d7cc9, 0x5005713c, 0x270241aa, 0xbe0b1010, 0xc90c2086, 0x5768b525, 0x206f85b3, 0xb966d409, 0xce61e49f, 0x5edef90e, 0x29d9c998, 0xb0d09822, 0xc7d7a8b4, 0x59b33d17, 0x2eb40d81, 0xb7bd5c3b, 0xc0ba6cad, 0xedb88320, 0x9abfb3b6, 0x03b6e20c, 0x74b1d29a, 0xead54739, 0x9dd277af, 0x04db2615, 0x73dc1683, 0xe3630b12, 0x94643b84, 0x0d6d6a3e, 0x7a6a5aa8, 0xe40ecf0b, 0x9309ff9d, 0x0a00ae27, 0x7d079eb1, 0xf00f9344, 0x8708a3d2, 0x1e01f268, 0x6906c2fe, 0xf762575d, 0x806567cb, 0x196c3671, 0x6e6b06e7, 0xfed41b76, 0x89d32be0, 0x10da7a5a, 0x67dd4acc, 0xf9b9df6f, 0x8ebeeff9, 0x17b7be43, 0x60b08ed5, 0xd6d6a3e8, 0xa1d1937e, 0x38d8c2c4, 0x4fdff252, 0xd1bb67f1, 0xa6bc5767, 0x3fb506dd, 0x48b2364b, 0xd80d2bda, 0xaf0a1b4c, 0x36034af6, 0x41047a60, 0xdf60efc3, 0xa867df55, 0x316e8eef, 0x4669be79, 0xcb61b38c, 0xbc66831a, 0x256fd2a0, 0x5268e236, 0xcc0c7795, 0xbb0b4703, 0x220216b9, 0x5505262f, 0xc5ba3bbe, 0xb2bd0b28, 0x2bb45a92, 0x5cb36a04, 0xc2d7ffa7, 0xb5d0cf31, 0x2cd99e8b, 0x5bdeae1d, 0x9b64c2b0, 0xec63f226, 0x756aa39c, 0x026d930a, 0x9c0906a9, 0xeb0e363f, 0x72076785, 0x05005713, 0x95bf4a82, 0xe2b87a14, 0x7bb12bae, 0x0cb61b38, 0x92d28e9b, 0xe5d5be0d, 0x7cdcefb7, 0x0bdbdf21, 0x86d3d2d4, 0xf1d4e242, 0x68ddb3f8, 0x1fda836e, 0x81be16cd, 0xf6b9265b, 0x6fb077e1, 0x18b74777, 0x88085ae6, 0xff0f6a70, 0x66063bca, 0x11010b5c, 0x8f659eff, 0xf862ae69, 0x616bffd3, 0x166ccf45, 0xa00ae278, 0xd70dd2ee, 0x4e048354, 0x3903b3c2, 0xa7672661, 0xd06016f7, 0x4969474d, 0x3e6e77db, 0xaed16a4a, 0xd9d65adc, 0x40df0b66, 0x37d83bf0, 0xa9bcae53, 0xdebb9ec5, 0x47b2cf7f, 0x30b5ffe9, 0xbdbdf21c, 0xcabac28a, 0x53b39330, 0x24b4a3a6, 0xbad03605, 0xcdd70693, 0x54de5729, 0x23d967bf, 0xb3667a2e, 0xc4614ab8, 0x5d681b02, 0x2a6f2b94, 0xb40bbe37, 0xc30c8ea1, 0x5a05df1b, 0x2d02ef8d, }; uint32_t Crc32IEEE(const std::string data) { uint32_t crc 0xFFFFFFFF; for (char c : data) { crc Crc32IEEETable[(crc ^ static_castuint8_t(c)) 0xFF] ^ (crc 8); } return crc ^ 0xFFFFFFFF; }添加节点调用的其实就是AddNode添加结点是需要写入操作的所以我们用的是写锁然后遍历所有我们需要添加的结点并且同时我们需要把配置信息里面的当前虚拟结点数量传入进去因为新添加一个真实结点就需要添加对应的虚拟节点而新添加的虚拟节点我们会先对这个结点起一个有规律的名字我们通过自定义的Crc转化为int类型放入vector容器里面可能最后的容器里面是1 3 2 4 6 5 这样子没有顺序的样子所以我们需要重排而重排的目的就是为了之后得到数据的时候可以通过二分查找来快速找到key值bool ConsistentHashMap::Add(const std::vectorstd::string_view nodes) { if (nodes.empty()) { return false; } std::unique_lock lock{mutex_}; // 写锁 for (const auto node : nodes) { if (node.empty()) { continue; } AddNode(node, config_.replicas); } // 重排哈希环 std::sort(keys_.begin(), keys_.end()); return true; } void ConsistentHashMap::AddNode(const std::string_view node, int replicas) { for (int i 0; i replicas; i) { std::string hash_key fmt::format({}-{}, node, std::to_string(i)); uint32_t hash config_.hash_func(hash_key); keys_.push_back(hash); hash_map_[hash] node; } node_replicas_[node] replicas; // 如果这个结点是新添加的初始化其计数器 if (node_counts_.find(node) node_counts_.end()) { node_counts_[node] 0; } }移除结点就是要先删除虚拟结点然后删除对应的真实结点删除虚拟结点的办法就是根据我们之前的规律找到对应的key然后将其删除。我们这里使用的是remove erase的操作首先remove并不是删除其作用就是移动元素当执行完remove之后会返回一个指向新数组尾部的一个指针我们就可以利用这个指针来删除原来旧容器里面那些没有用的数据所以我们需要remove erase最后我们需要删除真实结点nodebool ConsistentHashMap::Remove(const std::string_view node) // 传入的就是真实结点 { if (node.empty()) { return false; } std::unique_lock lock{mutex_}; auto it_replicas node_replicas_.find(node); // 找虚拟映射结点 if (it_replicas node_replicas_.end()) { return false; } int replicas it_replicas-second; // 移除结点的所有虚拟节点 for (int i 0; i replicas; i) { std::string hash_key fmt::format({}-{}, node.data(), std::to_string(i)); uint32_t hash config_.hash_func(hash_key); hash_map_.erase(hash); auto it std::remove(keys_.begin(), keys_.end(), hash); keys_.erase(it, keys_.end()); } // 删除真实结点 node_replicas_.erase(node); node_counts_.erase(node); return true; }我们要查询数据的时候用二分查找这也是我们为什么之前需要对keys进行排序并且一定要记住我们一致性哈希是一个环所以当我们找到末尾的时候其实就是开头。然后我们通过虚拟结点找到对应的真实结点auto ConsistentHashMap::Get(const std::string_view key) - std::string_view { if (key.empty()) { return ; } if (keys_.empty()) { return ; } uint32_t hash config_.hash_func(key); // 二分查找 auto it std::lower_bound(keys_.begin(), keys_.end(), hash); // 处理边界情况 : 如果到了末尾则回到开头 if (it keys_.end()) { it keys_.begin(); } std::string_view node hash_map_[*it]; // 增加结点计数和总数请求数 node_counts_[node]; total_requests_; return node; }统计负载信息然后返回一个数组这个数组里面存储的是每一个结点的负载信息auto ConsistentHashMap::GetStarts() - std::unordered_mapstd::string, double { std::shared_lock lock{mutex_}; std::unordered_mapstd::string, double stats; long long curr_total total_requests_.load(); if (curr_total 0) { return stats; } for (auto const [node, count] : node_counts_) { stats[node.data()] static_castdouble(count.load()) / static_castdouble(curr_total); } return stats; }开启负载均衡其实就是创建一个新的线程这个线程里面专门执行负载均衡的检查和修正首先是这个检查其实就是计算当前结点里面最高的差异diff如果这个diff比我们设置的threadshold要大那么我们就应该修正了void ConsistentHashMap::StartBalancer() { is_balancer_stop_ false; balancer_thread_ std::thread{ [this] { while(!is_balancer_stop_) { std::this_thread::sleep_for(std::chrono::seconds(1)); if (!is_balancer_stop_) // 再次检查防止在sleep期间被要求停止 { CheckAndRebalance(); } } }}; } void ConsistentHashMap::CheckAndRebalance() { if (total_requests_.load() 1000) { return; // 样本太少了不进行调整 } std::shared_lock lock{mutex_}; if (node_replicas_.empty()) { return; } // 计算系统平均负载总请求数 / 物理节点数 long long current_total_requests total_requests_.load(); // 每台真实服务器的平均请求数用来判断哪台服务器负载过高 double avg_load static_castdouble(current_total_requests) / node_replicas_.size(); double max_diff 0.0; for (auto const [node, count] : node_counts_) { double diff std::abs(static_castdouble(count.load()) - avg_load); if (avg_load 0) { if (diff / avg_load max_diff) { max_diff diff / avg_load; } } else if (diff 0) // 如果avg是0但是counts不是0说明不平衡 { max_diff 1.0; } } lock.unlock(); // 释放读锁因为rebalenceNodes要写锁 if (max_diff config_.load_balance_threadshold) { RebalanceNodes(); } }这个就是平衡的措施这个平衡措施不是针对某一个结点而是所有的结点移除全部的结点之后再重新添加结点但是因为AddNode里面没有排序的措施所以我们还需要补充sort对于keys的排序void ConsistentHashMap::RebalanceNodes() { std::unique_lock lock{mutex_}; if (node_replicas_.empty()) { return; } long long current_total_requests total_requests_.load(); double avg_load static_castdouble(current_total_requests) / node_replicas_.size(); // 调整每个结点的虚拟结点数量 // 注意这里需要创建一个副本因为在循环里面可能会修改 nodeReplicas 和 nodeCounts std::unordered_mapstd::string_view, int curr_replicas node_replicas_; std::unordered_mapstd::string_view, std::atomiclong long curr_counts; for (auto const [node, count] : node_counts_) { curr_counts[node] count.load(); } for (auto const [node, count] : curr_counts) { int old_replicas curr_replicas[node]; double load_ratio 0.0; if (avg_load 0) { load_ratio static_castdouble(count) / avg_load; } else if (count 0) { load_ratio 2.0; } else { load_ratio 1.0; } int new_replicas; if (load_ratio 1.0) { // 负载过高减少虚拟结点 new_replicas static_castint(std::round(static_castdouble(old_replicas) / load_ratio)); } else { new_replicas static_castint(std::round(static_castdouble(old_replicas) * (2.0 - load_ratio))); } // 确保在有限的范围之内 if (new_replicas config_.min_replicas) { new_replicas config_.min_replicas; } else if (new_replicas config_.max_replicas) { new_replicas config_.max_replicas; } if (new_replicas ! old_replicas) { // 重新添加结点的虚拟节点先移除旧的结点再添加新的 // 移除结点的所有虚拟结点 int replicas_to_remove node_replicas_[node]; for (int i 0; i replicas_to_remove; i) { std::string hashKey static_caststd::string(node) - std::to_string(i); uint32_t hash config_.hash_func(hashKey); hash_map_.erase(hash); auto it std::remove(keys_.begin(), keys_.end(), hash); keys_.erase(it, keys_.end()); } node_replicas_.erase(node); AddNode(node, new_replicas); } } // 重置计数器 for (auto pair : node_counts_) { pair.second.store(0); } total_requests_.store(0); std::sort(keys_.begin(), keys_.end()); }本篇文章到这里就结束了希望可以帮助大家理解~~~