分布式哈希表原理與實現（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)