如何用java实现一个p2p种子搜索(2)-路由表实现

路由表实现

回顾一下上一篇讲的内容，上一篇提到从dht网络中获取infohash，那么加入dht网络后的最重要的第一步就是怎么去建立路由表。

路由表里面保存的是dht中其他node的信息，所以node可以这么设计

public class Node implements Comparable<Node>{ private String nodeId;//16进制字符串 private String ip; //node的ip private Integer port; //node的端口 private Date updateTime;//最后更新时间 private byte[] nodeIdBytes;//20字节 private Integer k=0;//k桶应该有的位置 private Integer currentK=0;//当前的位置 private Integer rank=0; //node的rank分值 ，路由表满的时候，优先移除分值低的
.....
}


因为路由表的每个bucket最多只有存8个，所以当路由表的bucket满的时候，需要不断的删除rank分最低的node，为了高效比较和删除bucket我们可以用PriorityQueue，每个路由表最多有160个bucket，所以可以用map来存储路由表

private Map<Integer,PriorityQueue<Node>> tableMap=new ConcurrentHashMap<>();


因为路由表一开始只有一个bucket，当节点数量超过8个就会分裂成两个bucket，为了确定新节点应该插入到哪个bucket中，所以把每个bucket设计成链表

public static class Bucket{
private int k; //当前是第几个k桶
private Bucket next;//下一个k桶
}


好了我们再来看怎么添加一个node

public void put(Node node) {
int bucketIndex = getBucketIndex(node);
if(bucketIndex==0){//是自己就不用加入了
return;
}
PriorityQueue<Node> pq = tableMap.get(bucketIndex);
if(CollectionUtils.isEmpty(pq)){
//如果是空 那么找最近的那个节点加入
boolean isAdd=false;
while(bucket.next != null){
if(bucketIndex > bucket.getK()
&& bucketIndex < bucket.next.getK()){
//先往小的里面放
node.setCurrentK(bucket.getK());
isAdd=putAccurate(tableMap.get(bucket.getK()),node,false,bucket,tableMap);
if(!isAdd){
node.setCurrentK(bucket.next.getK());
isAdd=putAccurate(tableMap.get(bucket.next.getK()),node,true,bucket,tableMap);
}
}
bucket=bucket.next; }
if(!isAdd){
//没有添加成功 那么往最后一个节点添加
node.setCurrentK(bucket.getK());
putAccurate(tableMap.get(bucket.getK()),node,true,bucket,tableMap);
} }else{//如果不空 那么直接加 简单点来吧
if(pq.size()<8){
if(!pq.contains(node)){
node.setCurrentK(node.getK());
pq.add(node);
}else{
reAdd(pq,node);
}
}else{
pq.add(node);
pq.poll();
}
}
}


其中比较重要的是方法是putAccurate

/**
* @param pq 当前bucket
* @param node 需要插入的node
* @param isSplit 是否需要分裂
* @param bucket 需要插入的bucket的位置
* @param tableMap 路由表
* @return 返回是否添加成功
*/
@SneakyThrows
public boolean putAccurate(PriorityQueue<Node> pq,Node node,boolean isSplit,Bucket bucket,Map<Integer,PriorityQueue<Node>> tableMap){
boolean isAdd=false;
if(pq.contains(node)){
return reAdd(pq,node);
}
if(pq.size()<8){
pq.add(node);
isAdd=true;
}
if(isSplit && !isAdd){
PriorityQueue<Node> priorityQueue=new PriorityQueue<Node>((x,y)->x.getRank()-y.getRank());
priorityQueue.add(node);
tableMap.putIfAbsent(node.getK(),priorityQueue);
//创建新的k桶后需要把两边的bucket距离比较近的都放到自己的k桶里面 如果超过8个就丢了 最好是可以ping一下
//先从小的开始放
PriorityQueue<Node> collect1 = new PriorityQueue<>();
collect1.addAll(tableMap.get(bucket.getK()).stream().filter(n -> {
if (priorityQueue.size() < 8 &&
Math.abs(n.getK() - n.getCurrentK()) > Math.abs(n.getK() - node.getK())) {
n.setCurrentK(node.getK());
priorityQueue.add(n);
return false;
}
return true;
}).collect(Collectors.toSet()));
tableMap.put(bucket.getK(),CollectionUtils.isNotEmpty(collect1)?collect1:new PriorityQueue<Node>());
if(bucket.next!=null && CollectionUtils.isNotEmpty(tableMap.get(bucket.next.getK()))){
PriorityQueue<Node> collect = new PriorityQueue<>();
collect.addAll(tableMap.get(bucket.next.getK()).stream().filter(n -> {
if (priorityQueue.size() < 8 &&
Math.abs(n.getK() - n.getCurrentK()) > Math.abs(n.getK() - node.getK())) {
n.setCurrentK(node.getK());
priorityQueue.add(n);
return false;
}
return true;
}).collect(Collectors.toSet()));
tableMap.put(bucket.next.getK(),CollectionUtils.isNotEmpty(collect)?collect:new PriorityQueue<Node>());
}
Bucket b=new Bucket(node.getK(),bucket.next);
bucket.next=b;
isAdd=true;
node.setCurrentK(node.getK());
}
return isAdd;
}


上一篇我们知道路由表主要通过find_node来建立，那我们自己也会收到别人发起的find_node请求，所以我们还要实现根据nodeid来查找最近的8个node

/**
* 根据nodeid 查找最近的8个node
* @param trargetBytes 需要查找目标id
* @return
*/
public List<Node> getForTop8(byte[] trargetBytes){
int bucketIndex = getBucketIndex(trargetBytes);
List<Node> l=new ArrayList<>();
PriorityQueue<Node> pq = tableMap.get(bucketIndex);
if(CollectionUtils.isEmpty(pq)){
while(bucket.next != null){
if(bucketIndex > bucket.getK()
&& bucketIndex < bucket.next.getK()){ tableMap.get(bucket.next.getK()).stream().forEach(x->{
if(l.size()<8){
l.add(x);
}
});
}
bucket=bucket.next;
}
if(CollectionUtils.isEmpty(l)){
tableMap.get(bucket.getK()).stream().forEach(x->{
if(l.size()<8){
l.add(x);
}
});
} }else{//如果不空 那么直接加 简单点来吧
l.addAll(pq.stream().collect(Collectors.toList()));
}
return l;
}


好了，到了这里路由表大致就实现啦。已经成功完成了第一步，现在呢路由表还没有初始化刚开始什么数据都没有，而且我们还是不能从dht中获取infohash，下一篇再来讲dht 协议，里面还会讲怎么初始化路由表，实现了dht协议也就完成了一大半了。

本章路由表部分还可以参考源码里面的RoutingTable，应该都能看得懂，地址:https://github.com/mistletoe9527/dht-spider