分布式哈希表原理与实现（Python版和Java版）

 

 

本文介绍了分布式哈希表的概念及实现，这种描述非常适合代码模拟实现。文末附上了实现的代码（Java版和Python版）

本文内容（除代码外）皆来自 《Foundations of Computers Systems Research》一书，自己翻译的，转载请注明出处，不准确的部分请告知，欢迎讨论。

 

什么是分布式哈希表？

 

    分布式哈希表是一个基本的结构，被广泛的应用在许多分布式系统中，用于组织动态改变的（分布式）节点集合和透明的资源定位服务，如，DHT 在许多 P2P 系统中作为一个基本结构提供高效透明的资源定位服务。所有节点都同等重要，这使得 DHT 更加的装载平横。而且，在 DHT 中具有一定规律的拓扑结构（环、树、超立方体、网格、蝴蝶形……），这使得查找更加的高效。虽然，各种 DHT 实现的拓扑结构各不相同，但它们至少提供以下两个基本的功能：

• put(key, value): 插入一个键值对到 DHT 网络中，不需要知道具体的存储地点
• value = get(key)：根据 key 从 DHT 网络中检索相应的值，不需要知道具体的存储地点

 

DHT 网络设计的重点大多在以下几个方面：

• 拓扑结构（Topology structure）
• 路由协议（Routing protocol）
• 动态维护和失败恢复（Dynamic maintenance and failure recovery）

 

 

各种 DHT 对比

DHTs	Topology	Node Degree	Routing Hops
CAN	mesh	O(d)	O(dn^(1/d))
Chord	ring	O(logn)	O(logn)
Pastry	tree + ring	O(logn)	O(logn)
Tapestry	tree	O(logn)	O(logn)
Kademlia	tree	O(logn)	O(logn)
Viceroy	butterfly	7	O(logn)
Koord	de Bruijn graph + ring	2 or O(logn)	O(logn) or O(logn/loglogn)
Cycloid	hypecube + ring	3	O(d) with n = d2^d

 

一个 DHT 例子 —— Chord

Chord 的拓扑结构 Chord 使用一致性哈希（consistent hashing）将键(key)和节点(node)映射成一个 m 位的标识符 Chord 使用环形作为它的拓扑结构，节点按照顺时针，标识符递增排列。因此，每一个节点都有一个直接前驱和一个直接后继。 每一个节点负责存储标识符范围为（i-1, i] 的键。i 为节点的标识符，i-1 为当前节点前驱节点的标识符。 拓扑结构如右图。N14 节点负责存储标识符为（9，10，11，12，13，14）的键
Chord 的查找算法     若是使用普通的遍历查找，那么每个节点的度为1（只有后继），查找的时间复杂度为log(n)。而 chord 在节点度数和时间复杂度之间做了个折中：增加节点的度，减少查找时间。     每个节点都有一个 finger table，该表有 m 个条目，每个条目指向另外的节点。条目指向的节点计算公式为：(n + 2^k) % 2^m 。（n 为当前节点的标识符，m 是标识符空间的位数，0 <= k < m）。如右图。     为什么要设置这样一个指向表呢？为了更快的查找。假设现在我们要从节点 n 开始，查找标识符 id。于是有如下的步骤： 计算当前节点 n 到目标节点(存储 id)的距离：d = (id-n+2^m) % 2^m，从这里可以知道:（n + d) % 2^m = id 然后计算最大的 k，使得 2^k <= d。然后根据 finger table，跳转到存储标识符 n + 2^k 的节点。现在离目标又近了一步 以上过程递归执行，直到目标标识符位于当前节点和其直接后继之间，那么此时目标标识符就存储在当前节点的直接后继中
Chord 的动态维护和失败节点恢复     在分布式系统中，会经常的进行节点的动态加入和离开，而且节点也可能出问题，所以还需要考虑节点的动态维护，一些算法就要周期的执行来维护拓扑结构。举一个例子，节点 N26 加入该网络： 节点 N26 向节点 n 请求加入网络 n 会找到为标识符 26 负责的节点 N32，N32 就是 N26 的直接后继 节点 N26 会修改自己的后继，指向 N32，然后执行一个算法更新后继，并且给新后继 N32 发送一条问候信息 N32 收到该消息后，发现 N26 离自己比当前前驱 N21 更近，所以，N32 就会重设它的前驱为 N26 N21 执行一个算法更新后继，发现 N26 是自己的新后继，于是修改指向，并发送问候信息给 N26 N26 收到消息后，发现自己的前驱为空，然后重设自己的前驱为 N21 上面的过程能够很好的处理节点的加入，但当节点出了问题时，就很难得知其后继节点信息，因此，Chord 还维护了一张后继信息表，但不是记录所有节点的，表长为 r = log(N)，当节点出问题后，能够有较大概率获取正确的后继节点信息。

 

 

 

学习案例

案例一：协同域名系统（Cooperative Domain Name System，CoDoNS）

    CoDoNS 是一种建立在 DHT 之上的自组织 P2P 网络，它被布置在域名系统客户端（DNS解析模块，存在于终端设备中）与旧版 DNS 服务器之间，用来处理旧版 DNS 服务器的性能瓶颈。 传统的 DNS 服务器具有如下的性能瓶颈（缺点）：

1. 抵御 DoS 攻击能力差。服务器负载高，研究发现，有80%的域名仅被两个域名服务器服务；链路负载高，研究发现，有30%的域名仅仅通过单一的路由链路与互联网链接。这些问题很容易被恶意攻击者利用。
2. 域名，地址转换具有很高的延迟
3. 其缓存策略不允许它进行快速的更新传播，而且可能不能快速反应突发变化

        CoNoNS 结构示意图

 

                          

在CoDoNS 系统中，查询及回应的流程如下：

1. 客户端发送一个查询请求到本地 CoDoNS 服务器，若本地 CoDoNS 服务器有该域名的缓存备份，就立即返回，否则执行下一步
2. 该查询请求被传送到一个中间 CoDoNS 服务器，同上，如该服务器有域名备份，就直接返回，否则执行下一步
3. 中间 CoDoNS 服务器将该请求发送到 home node（也是一个 CoDoNS 服务器），若 home node 没有该记录，就从传统 DNS 域名系统检索该记录
4. home node 检索到记录后，直接将该信息返回到最初的 CoDoNS 服务器（即最开始的本地 CoDoNS 服务器），然后由它返回给客户端。与此同时，home node 将该记录插入到自己的缓存中，然后传播该记录给邻居节点（基于某种优先级）。

 

home node 是指：其标识符最接近（在所有的 CoDoNS 服务器中）于某一个域名经过一致性哈希映射后的值。也就是说，home node 不是 一定的，它会随着请求查询的域名的不同而动态分配。

 

案例二：协同文件系统（Cooperative File System，CFS）

    一个理想的 P2P 系统应该具有高性能、容错性、加载平衡、适应于突发的变化等特点。CFS 采用是分布式架构，具有如上的优点。它采用了分布式哈希块存储系统（DHash block storage system）和 Chord DHT，从而提供了一种细粒度（fine granular）的存储服务。
    CFS 是一种只读的块级文件系统，其设计的关键之处在于，将每个文件划分成多个块，并且将它们分布在多个服务器中，以实现加载平衡。由于现实生活中，各个服务器的存储容量和网络带宽各不相同，因此，CFS 常常将一个实际的服务器划分成多个虚拟服务器，虚拟服务器的数量由该实际服务器的容量和带宽决定。
    CFS 由两层组成，上层为 DHash，它负责将文件块分布到服务器中，维护合适的备份水平，缓存热点文件以加速寻找。下层是 Chord 层，它负责定位具体块的存储服务器。Chord 中的每一个节点都维护了两个数据结构：后继列表和finger table。后继列表是为了在节点出问题以后依然可以保持链接，finger table 是为了提供高效的查找定位服务。

 

 

 

代码实现

Java（包含四个文件：测试类、节点类、Chord类、节点IP地址文件）

//只写了初始化数据结构、节点加入、节点查询
//没有写节点的移除，数据的存储

IPAddressList.txt
211.138.116.246
128.001.223.111
176.002.222.123
223.104.160.069
112.017.247.082
203.069.010.164



​package DisturbedHashTable;

import java.io.IOException;
//测试类
public class MainTest {
    public static void main  (String[] args) throws IOException{
        ChordDHT dht = new ChordDHT(6, "F:\\VSCodeProjects\\JavaProjects\\DisturbedHashTable\\IPAddressList.txt");
        dht.printNode();

        //测试加入
        Node node = new Node(6, "242.108.11.31");
        Node curNode = dht.getNode(32);
        dht.joinToRing(curNode, node, 6);

        var tmp = node.getSuccessor();
        System.out.printf("\n%d ", node.getIdentifier());
        while(tmp != node){
            System.out.printf("%d ", tmp.getIdentifier());
            tmp = tmp.getSuccessor();
        }

    }
}

//Chord类
package DisturbedHashTable;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.HashMap;
import java.util.Scanner;

/**
 * 基本信息： 
 * 1. Chord 是一个环状结构 
 * 2. 每个节点、键值都被映射成了 m 位的标识符
 * 3. 支持查找、插入、移除节点操作 
 * 4. 每个节点负责存储标识符为 （i-1, i] 5. 每个节点都有一个 fingerTable
 * 
 * 实现想法： 
 * 1. 利用双向循环链表作为环状结构 (X) 不用链表，链表不能随机访问，用一个 HashMap 
 * 2. 标识符空间设置位 6 位 
 * 3. 哈希函数将各个节点的IP地址映射成为 6 位的标识符 
 * 4. 每个节点还有一个链表存储其负责的标识符和 fingerTable
 * 
 */

 public class ChordDHT{
     //节点集合，键值是其标识符
    private HashMap<Integer, Node> nodeSet;  
    /**
       * 构造函数接收两个参数：
       *    m        标识符空间位数
       *    srcfile 节点IP地址列表文件
       *
       */
      public ChordDHT(int m, String srcfile) throws IOException{
          this.nodeSet = new HashMap<Integer, Node>();
          Scanner inFile = new Scanner(Path.of(srcfile), StandardCharsets.UTF_8);
          Node firstNode = null;

          while(inFile.hasNext() == true){ //加入初始节点
              String ip = inFile.nextLine();
              Node node = new Node(m, ip);
              if(firstNode == null) firstNode = node;
              this.joinToRing(firstNode, node, m);
          }
          inFile.close();
          
      }


      /**
       * 处理新节点的加入
       * 除了改变前驱后继关系，还可能存在key的迁移，
       * 因为这可能会改变某个节点负责的标识符范围
       * @param curNode 当前节点 (受理加入请求的节点)
       * @param id 请求加入的节点
       * @param m 标识符空间位数
       * @return 是否加入成功
       */
      public boolean joinToRing(Node curNode, Node id, int m){
            
           //环不为空时 找到id的直接后继, 并设置相应的域
          if(this.nodeSet.isEmpty() == false){
              var successor = this.searchNode(curNode, id, m);
              var predecessor = successor.getPredecessor();
              id.setPredecessor(predecessor);
              id.setSuccessor(successor);
              successor.setPredecessor(id);
              predecessor.setSuccessor(id);
          }
          //将节点放入环中, 包含了环为空的情况
          this.nodeSet.put(id.getIdentifier(), id);
          
          //节点加入后，可能会改变一部分节点的fingerTable
          for(var node: this.nodeSet.values()){
              this.updateFingerTable(node, m);
          }  
          return true; 
      }

      public void updateFingerTable(Node node, int m){
          for(int i = 0; i < m; i++){ //对表的每一个条目
              var aIdentifier = ((int)Math.pow(2, i) + node.getIdentifier()) % (int)Math.pow(2, m);
              
              //找到为此标识符负责的节点
              for(var item: this.nodeSet.values()){
                  if(item.inStorageBound(aIdentifier, m) == true){
                      node.getFingerTable().put((int)Math.pow(2, i), item);
                      break;
                  }
              }
          }
      }


      public Node searchNode(Node curNode, Node id, int m){
          //System.out.println("test searchNode");
          Node result = null;
          //出口
          //负责id标识符的节点必是id的后继节点
          if(curNode.inStorageBound(id.getIdentifier(), m) == true){
              result = curNode; 
          }
          
          // 递归情况
          else{
              var dis = (id.getIdentifier() - curNode.getIdentifier() + (int)Math.pow(2, m)) % (int)Math.pow(2, m);
              //找到最大的 k 使得 2^k <= dis
              var maxk = 0;
              while((int)Math.pow(2, maxk) <= dis) maxk++;
              maxk--;
              result = searchNode(curNode.getFingerTable().get((int)Math.pow(2, maxk)), id, m);
          }
          return result;
      }

      public boolean isIn(int key){
          return nodeSet.containsKey(key);
      }

      public Node getNode(int key){
          return nodeSet.get(key);
      }

      public void printNode(){
          for(var node: this.nodeSet.values()){
              System.out.printf("identifier is %d, pre is %d, suc is %d\n",
              node.getIdentifier(), node.getPredecessor().getIdentifier(), 
              node.getSuccessor().getIdentifier());
          }
      }
 }

//节点类
package DisturbedHashTable;

import java.sql.Time;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Random;

public class Node {
    private int identifier;
    private HashMap<Integer, Node> fingerTable; //键值为 1，2，4，8，16……
    private HashMap<Integer, String> keyList; //存储其负责的标识符，String 是 key
    private Node predecessor; //前驱
    private Node successor; //后继

    public Node(int m, String ip){
        //将ip地址映射成m位标识符，生成m个表项目
        this.identifier = hashFunc(m, ip);
        this.keyList = new HashMap<Integer, String>();
        this.fingerTable = new HashMap<Integer, Node>();
        for(int i = 0; i < m; i++){ 
            this.fingerTable.put((int)Math.pow(2, i), this);
        }
        this.predecessor = this;
        this.successor = this;
    }

    private int hashFunc(int m, String ip){
        Random rd = new Random(System.currentTimeMillis());
        int result = 0;//rd.nextInt(999999); //增加随机数，减少碰撞概率
        for(int i = 0; i < ip.length(); i++){
            if(ip.charAt(i) <= '9' && ip.charAt(i) >= '0') {
                result = (result + (int)ip.charAt(i))  % (int)Math.pow(2, m);
            }
        }
        return result % (int)Math.pow(2, m);
    }

    public HashMap<Integer, Node> getFingerTable(){
        return this.fingerTable;
    }

    

    //负责标识符的范围
    /**
      * 这里可能有三种可能：
      * 1. 前驱节点标识符更大（如，23 ---> 1）
      * 2. 前驱节点标识符更小（如，23 ---> 27）
      * 3. 前驱节点标识符一样大（环中只有一个节点时）
      */
    public boolean inStorageBound(int aIdentifier, int m){
        var lower = this.predecessor.identifier;
        var upper = this.identifier;

        //第一种情况
        if(lower > upper){
            if(aIdentifier > lower && aIdentifier < (int)Math.pow(2, m)){
                return true;
            }
            if(aIdentifier < upper){
                return true;
            }
        }

        //第二种情况
        if(aIdentifier > lower && aIdentifier <= upper){
            return true;
        }

        //第三种情况
        //由于只有一个节点，所以它负责所有的标识符
        if(lower == upper){
            return true;
        }

        return false;
        
    }

    public int getIdentifier(){
        return this.identifier;
    }

    public Node getPredecessor(){
        return this.predecessor;
    }

    public Node getSuccessor(){
        return this.successor;
    }

    public boolean setPredecessor(Node node){
        this.predecessor = node;
        return true;
    }

    public boolean setSuccessor(Node node){
        this.successor = node;
        return true;
    }

}

 

Python（都写在一个文件中，节点IP地址文件同上）

 

import random
import time


ringLenth = lambda x: int(pow(2, x))


class Node:
    ''' 哈希表中的节点数据结构 '''
    
    def __init__(self, ip, m):
        self.id = self.generateId(ip = ip, m = m)
        self.fingerTable = self.generateFT(m = m)
        self.predecessor = self
        self.sucessor = self


    def generateId(self, ip, m):
        result = 0
        random.seed(time.time())
        for ch in ip:
            result = (result + ord(ch) + random.randint(1, 99)) % ringLenth(m)
        return result


    def generateFT(self, m):
        result = {}
        for i in range(m):
            result[int(pow(2, i))] = self
        return result


    def updateFT(self, m, dht):
        for item in self.fingerTable:
            pointer = (item + self.id) % ringLenth(m)
            for node in dht.getNodes():
                if node.isInBound(pointer, m):
                    self.fingerTable[item] = node
                    break
        

    def isInBound(self, pointer, m):
        # 每个节点负责的标识符范围，有三种情况
        # case 1 环中只有一个节点时
        if self.id == self.predecessor.getId():
            return True
        # case 2 前驱节点标识符较小
        elif self.id > self.predecessor.getId():
            if pointer > self.predecessor.getId() and pointer <= self.id:
                return True
        # case 3 前驱节点标识符较大
        elif self.id < self.predecessor.getId():
            if pointer > self.predecessor.getId() and pointer < ringLenth(m):
                return True
            if pointer < self.id:
                return True
        
        else:
             return False

    def getSucessor(self):
        return self.sucessor

    def getPredecessor(self):
        return self.predecessor

    def setSuceesor(self, node):
        self.sucessor = node
    
    def setPredecessor(self, node):
        self.predecessor = node

    def getId(self):
        return self.id 


    def getFT(self):
        return self.fingerTable
 



class DHT:
    ''' 哈希表数据结构 '''
    def __init__(self, fileName, m):
        self.nodeSet = {}
        firstNode = None
        for ip in open(fileName, 'r').readlines():
            node = Node(ip, m)
            if firstNode is None:
                firstNode = node
            self.joinToRing(firstNode, node, m)


    def joinToRing(self, curNode, wnode, m):
        if len(self.nodeSet) is not 0:
            sucessor = self.searchSucessor(curNode, wnode, m)
            predecessor = sucessor.getPredecessor()

            wnode.setPredecessor(predecessor)
            wnode.setSuceesor(sucessor)
            sucessor.setPredecessor(wnode)
            predecessor.setSuceesor(wnode)
        
        self.nodeSet[wnode.getId()] = wnode
        self.updateNodesFT(m)


    def searchSucessor(self, curNode, wnode, m):
        if curNode.isInBound(wnode.getId(), m):
            return curNode
        else:
            dis = (wnode.getId() - curNode.getId() + ringLenth(m)) % ringLenth(m)
            # 寻找最大的 k 使得 2^k <= dis
            maxk = 0
            while((2 ** maxk) <= dis): maxk += 1
            maxk -= 1
            return self.searchSucessor(curNode.getFT()[2 ** maxk], wnode, m)


    def updateNodesFT(self, m):
        for node in self.nodeSet:
            self.nodeSet[node].updateFT(m, self)

    def getNodes(self):
        return self.nodeSet.values()




class Test:
    ''' 测试哈希表是否运行正常 ''' 
    def __init__(self, fileName, m):
        self.dht = DHT(fileName, m) 
        for node in self.dht.getNodes():
            print("I am node %d, My predecessor is node %d, and My sucessor is node %d"
            % (node.getId(), node.getPredecessor().getId(), node.getSucessor().getId()))



if __name__ == "__main__":
    test = Test("DistributedHashTable\\IPAddressList.txt", 6)