DHT是什么

DHT全称叫分布式哈希表(Distributed Hash Table)，是一种分布式存储方法。
在不需要服务器的情况下，每个客户端负责一个小范围的路由，并负责存储一小部分数据
从而实现整个DHT网络的寻址和存储

Kademlia协议实现

要加入一个DHT网络，需要首先知道这个网络中的任意一个节点。
如何获得这个节点？在一些开源的P2P软件中，会提供一些节点地址
主要协议

  • ping(用于确定某个节点是否在线。这个请求主要用于辅助路由表的更新)
  • find_node(用于查找某个节点，以获得其地址信息。)
  • get_peer(通过资源的infohash获得资源对应的peer列表。)

  • announce_peer(通知其他节点自己开始下载某个资源,announce_peer中会携带get_peer回应消息里的token。)

    工作原理

  • 通过其他节点的announce_peer发来的infohash确认网络中有某个资源可被下载

  • 通过从网络中获取这个资源的种子文件，来获得该资源的描述
  • 不停的认识新节点,让远程节点保存自身到远程的路由表中

    工作流程

  1. BOOTSTRAP过程,加入DHT网络(主动认识DHT网络的其中一个节点)
  2. 加入进DHT网络后。远端节点会主动告诉我们它认识哪些节点
  3. 认识远端节点认识的节点
  4. 当远端成功保存自身节点到远端路由表中的时候，目的达成
  5. 等待远端的announce_peer消息
  6. 成功获取远端的下载hash

    程序源码 | Python

    ```python

    coding: utf-8

import socket from hashlib import sha1 from random import randint from struct import unpack, pack from socket import inet_aton, inet_ntoa from bisect import bisect_left from threading import Timer

from time import sleep

from bencode import bencode, bdecode

BOOTSTRAP_NODES = [ (“router.bittorrent.com”, 6881), (“dht.transmissionbt.com”, 6881), (“router.utorrent.com”, 6881) ] TID_LENGTH = 4 KRPC_TIMEOUT = 10 REBORN_TIME = 5 * 60 K = 8

def entropy(bytes): s = “” for i in range(bytes): s += chr(randint(0, 255)) return s

  1. # """把爬虫"伪装"成正常node, 一个正常的node有ip, port, node ID三个属性, 因为是基于UDP协议,     
  2. # 所以向对方发送信息时, 即使没"明确"说明自己的ip和port时, 对方自然会知道你的ip和port,     
  3. # 反之亦然. 那么我们自身node就只需要生成一个node ID就行, 协议里说到node ID用sha1算法生成,     
  4. # sha1算法生成的值是长度是20 byte, 也就是20 * 8 = 160 bit, 正好如DHT协议里说的那范围: 0 至 2的160次方,     
  5. # 也就是总共能生成1461501637330902918203684832716283019655932542976个独一无二的node.     
  6. # ok, 由于sha1总是生成20 byte的值, 所以哪怕你写SHA1(20)或SHA1(19)或SHA1("I am a 2B")都可以,     
  7. # 只要保证大大降低与别人重复几率就行. 注意, node ID非十六进制,     
  8. # 也就是说非FF5C85FE1FDB933503999F9EB2EF59E4B0F51ECA这个样子, 即非hash.hexdigest(). """

def random_id(): hash = sha1() hash.update( entropy(20) ) return hash.digest()

def decode_nodes(nodes): n = [] length = len(nodes) if (length % 26) != 0: return n for i in range(0, length, 26): nid = nodes[i:i+20] ip = inet_ntoa(nodes[i+20:i+24]) port = unpack(“!H”, nodes[i+24:i+26])[0] n.append( (nid, ip, port) ) return n

def encode_nodes(nodes): strings = [] for node in nodes: s = “%s%s%s” % (node.nid, inet_aton(node.ip), pack(“!H”, node.port)) strings.append(s)

  1. return "".join(strings)

def intify(hstr):

  1. #"""这是一个小工具, 把一个node ID转换为数字. 后面会频繁用到.""" 
  2. return long(hstr.encode('hex'), 16) #先转换成16进制, 再变成数字

def timer(t, f): Timer(t, f).start()

class BucketFull(Exception): pass

class KRPC(object): def init(self): self.types = { “r”: self.response_received, “q”: self.query_received } self.actions = { “ping”: self.ping_received, “find_node”: self.find_node_received, “get_peers”: self.get_peers_received, “announce_peer”: self.announce_peer_received, }

  1.     self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  2.     self.socket.bind(("0.0.0.0", self.port))
  4. def find_node_handler(self,msg):
  5.     pass
  7. def response_received(self, msg, address):
  8.     self.find_node_handler(msg)
  10. def query_received(self, msg, address):
  11.     try:
  12.         self.actions[msg["q"]](msg, address)
  13.     except KeyError:
  14.         pass
  16. def send_krpc(self, msg, address):
  17.     try:
  18.         self.socket.sendto(bencode(msg), address)
  19.     except:
  20.         pass

class Client(KRPC): def init(self, table): self.table = table

  1.     timer(KRPC_TIMEOUT, self.timeout)
  2.     timer(REBORN_TIME, self.reborn)
  3.     KRPC.__init__(self)
  5. def find_node(self, address, nid=None):
  6.     print "find node:",address
  7.     nid = self.get_neighbor(nid) if nid else self.table.nid
  8.     tid = entropy(TID_LENGTH)
  10.     msg = {
  11.         "t": tid,
  12.         "y": "q",
  13.         "q": "find_node",
  14.         "a": {"id": nid, "target": random_id()}
  15.     }
  16.     self.send_krpc(msg, address)
  18. def find_node_handler(self, msg):
  19.     try:
  20.         nodes = decode_nodes(msg["r"]["nodes"])
  21.         for node in nodes:
  22.             (nid, ip, port) = node
  23.             if len(nid) != 20: continue
  24.             if nid == self.table.nid: continue
  25.             self.find_node( (ip, port), nid )
  26.     except KeyError:
  27.         pass
  29. def joinDHT(self):
  30.     for address in BOOTSTRAP_NODES: 
  31.         self.find_node(address)
  33. def timeout(self):
  34.     if len( self.table.buckets ) < 2:
  35.         self.joinDHT()
  36.     timer(KRPC_TIMEOUT, self.timeout)
  38. def reborn(self):
  39.     self.table.nid = random_id()
  40.     self.table.buckets = [ KBucket(0, 2**160) ]
  41.     timer(REBORN_TIME, self.reborn)
  43. def start(self):
  44.     self.joinDHT()
  46.     while True:
  47.         try:
  48.             (data, address) = self.socket.recvfrom(65536)
  49.             msg = bdecode(data)
  50.             self.types[msg["y"]](msg, address)
  51.         except Exception:
  52.             pass
  54. def get_neighbor(self, target):
  55.     return target[:10]+random_id()[10:]

class Server(Client): def init(self, master, table, port): self.table = table self.master = master self.port = port Client.init(self, table)

  1. def ping_received(self, msg, address):
  2.     try:
  3.         nid = msg["a"]["id"]
  4.         msg = {
  5.             "t": msg["t"],
  6.             "y": "r",
  7.             "r": {"id": self.get_neighbor(nid)}
  8.         }
  9.         self.send_krpc(msg, address)
  10.         self.find_node(address, nid)
  11.     except KeyError:
  12.         pass
  14. def find_node_received(self, msg, address):
  15.     try:
  16.         target = msg["a"]["target"]
  17.         neighbors = self.table.get_neighbors(target)
  19.         nid = msg["a"]["id"]
  20.         msg = {
  21.             "t": msg["t"],
  22.             "y": "r",
  23.             "r": {
  24.                 "id": self.get_neighbor(target), 
  25.                 "nodes": encode_nodes(neighbors)
  26.             }
  27.         }
  28.         self.table.append(KNode(nid, *address))
  29.         self.send_krpc(msg, address)
  30.         self.find_node(address, nid)
  31.     except KeyError:
  32.         pass
  34. def get_peers_received(self, msg, address):
  35.     try:
  36.         infohash = msg["a"]["info_hash"]
  38.         neighbors = self.table.get_neighbors(infohash)
  40.         nid = msg["a"]["id"]
  41.         msg = {
  42.             "t": msg["t"],
  43.             "y": "r",
  44.             "r": {
  45.                 "id": self.get_neighbor(infohash), 
  46.                 "nodes": encode_nodes(neighbors)
  47.             }
  48.         }
  49.         self.table.append(KNode(nid, *address))
  50.         self.send_krpc(msg, address)
  51.         self.master.log(infohash)
  52.         self.find_node(address, nid)
  53.     except KeyError:
  54.         pass
  56. def announce_peer_received(self, msg, address):
  57.     try:
  58.         infohash = msg["a"]["info_hash"]
  59.         nid = msg["a"]["id"]
  61.         msg = { 
  62.             "t": msg["t"],
  63.             "y": "r",
  64.             "r": {"id": self.get_neighbor(infohash)}
  65.         }
  67.         self.table.append(KNode(nid, *address))
  68.         self.send_krpc(msg, address)
  69.         self.master.log(infohash)
  70.         self.find_node(address, nid)
  71.     except KeyError:
  72.         pass

该类只实例化一次.

class KTable(object):

  1. # 这里的nid就是通过node_id()函数生成的自身node ID. 协议里说道, 每个路由表至少有一个bucket,

还规定第一个bucket的min=0, max=2^160次方, 所以这里就给予了一个buckets属性来存储bucket, 这个是列表.

  1. def __init__(self, nid):
  2.     self.nid = nid
  3.     self.buckets = [ KBucket(0, 2**160) ]
  5. def append(self, node):
  6.     index = self.bucket_index(node.nid)
  7.     try:
  8.         bucket = self.buckets[index]
  9.         bucket.append(node)
  10.     except IndexError:
  11.         return
  12.     except BucketFull:
  13.         if not bucket.in_range(self.nid): 
  14.             return
  15.         self.split_bucket(index)
  16.         self.append(node)
  19.     # 返回与目标node ID或infohash的最近K个node.    
  21.     # 定位出与目标node ID或infohash所在的bucket, 如果该bucuck有K个节点, 返回.     
  22.     # 如果不够到K个节点的话, 把该bucket前面的bucket和该bucket后面的bucket加起来, 只返回前K个节点.    
  23.     # 还是不到K个话, 再重复这个动作. 要注意不要超出最小和最大索引范围.    
  24.     # 总之, 不管你用什么算法, 想尽办法找出最近的K个节点.    
  25. def get_neighbors(self, target):
  26.     nodes = []
  27.     if len(self.buckets) == 0: return nodes
  28.     if len(target) != 20 : return nodes
  30.     index = self.bucket_index(target)
  31.     try:
  32.         nodes = self.buckets[index].nodes
  33.         min = index - 1
  34.         max = index + 1
  36.         while len(nodes) < K and ((min >= 0) or (max < len(self.buckets))):
  37.             if min >= 0:
  38.                 nodes.extend(self.buckets[min].nodes)
  40.             if max < len(self.buckets):
  41.                 nodes.extend(self.buckets[max].nodes)
  43.             min -= 1
  44.             max += 1
  46.         num = intify(target)
  47.         nodes.sort(lambda a, b, num=num: cmp(num^intify(a.nid), num^intify(b.nid)))
  48.         return nodes[:K] #K是个常量, K=8 
  49.     except IndexError:
  50.         return nodes
  52. def bucket_index(self, target):
  53.     return bisect_left(self.buckets, intify(target))
  56.     # 拆表    
  58.     # index是待拆分的bucket(old bucket)的所在索引值.     
  59.     # 假设这个old bucket的min:0, max:16. 拆分该old bucket的话, 分界点是8, 然后把old bucket的max改为8, min还是0.     
  60.     # 创建一个新的bucket, new bucket的min=8, max=16.    
  61.     # 然后根据的old bucket中的各个node的nid, 看看是属于哪个bucket的范围里, 就装到对应的bucket里.     
  62.     # 各回各家,各找各妈.    
  63.     # new bucket的所在索引值就在old bucket后面, 即index+1, 把新的bucket插入到路由表里. 
  64. def split_bucket(self, index):
  65.     old = self.buckets[index]
  66.     point = old.max - (old.max - old.min)/2
  67.     new = KBucket(point, old.max)
  68.     old.max = point
  69.     self.buckets.insert(index + 1, new)
  70.     for node in old.nodes[:]:
  71.         if new.in_range(node.nid):
  72.             new.append(node)
  73.             old.remove(node)
  75. def __iter__(self):
  76.     for bucket in self.buckets:
  77.         yield bucket

class KBucket(object): slots = (“min”, “max”, “nodes”)

  1. # min和max就是该bucket负责的范围, 比如该bucket的min:0, max:16的话,     
  2. # 那么存储的node的intify(nid)值均为: 0到15, 那16就不负责, 这16将会是该bucket后面的bucket的min值.     
  3. # nodes属性就是个列表, 存储node. last_accessed代表最后访问时间, 因为协议里说到,     
  4. # 当该bucket负责的node有请求, 回应操作; 删除node; 添加node; 更新node; 等这些操作时,     
  5. # 那么就要更新该bucket, 所以设置个last_accessed属性, 该属性标志着这个bucket的"新鲜程度". 用linux话来说, touch一下.    
  6. # 这个用来便于后面说的定时刷新路由表.    
  8. def __init__(self, min, max):
  9.     self.min = min
  10.     self.max = max
  11.     self.nodes = []
  14. # 添加node, 参数node是KNode实例.    
  16. # 如果新插入的node的nid属性长度不等于20, 终止.    
  17. # 如果满了, 抛出bucket已满的错误, 终止. 通知上层代码进行拆表.    
  18. # 如果未满, 先看看新插入的node是否已存在, 如果存在, 就替换掉, 不存在, 就添加,    
  19. # 添加/替换时, 更新该bucket的"新鲜程度".    
  20. def append(self, node):
  21.     if node in self:
  22.         self.remove(node)
  23.         self.nodes.append(node)
  24.     else:
  25.         if len(self) < K:
  26.             self.nodes.append(node)
  27.         else:
  28.             raise BucketFull
  30. def remove(self, node):
  31.     self.nodes.remove(node)
  33. def in_range(self, target):
  34.     return self.min <= intify(target) < self.max
  36. def __len__(self):
  37.     return len(self.nodes)
  39. def __contains__(self, node):
  40.     return node in self.nodes
  42. def __iter__(self):
  43.     for node in self.nodes:
  44.         yield node
  46. def __lt__(self, target):
  47.     return self.max <= target

class KNode(object):

  1.  # """    
  2. #        nid就是node ID的简写, 就不取id这么模糊的变量名了. __init__方法相当于别的OOP语言中的构造方法,     
  3. #        在python严格来说不是构造方法, 它是初始化, 不过, 功能差不多就行.    
  4. #        """ 
  5. __slots__ = ("nid", "ip", "port")
  7. def __init__(self, nid, ip, port):
  8.     self.nid = nid
  9.     self.ip = ip
  10.     self.port = port
  12. def __eq__(self, other):
  13.     return self.nid == other.nid

using example

class Master(object): def init(self, f): self.f = f self.hashArr = []

  1. def log(self, infohash):
  2.     nhash = infohash.encode("hex")
  3.     if nhash not in self.hashArr:
  4.         self.hashArr.append(nhash)
  5.         self.f.write(+"\n")
  6.         self.f.flush()

try: print “start DHT Spider” f = file(“hash.txt”,”a+”) m = Master(f) s = Server(Master(f), KTable(random_id()), 6881) s.start()
except KeyboardInterrupt: s.socket.close() f.close()

```

参考自