如何用Python的Kademlia库定期移除无效节点并刷新路由表？

问题背景 #

已通过pip install kademlia安装Kademlia DHT，当前部署3个节点：1个引导初始节点、Node2（执行set操作）、Node3（执行get操作）。需求是当某个非初始节点离线时，让其他节点自动移除该无效节点并刷新路由表。已知Kademlia包提供lonely_buckets()、remove_contact(node)和flush()方法，需明确具体实现方式及刷新操作的必要性。

现有节点代码 #

初始节点代码

import logging
import asyncio

from kademlia.network import Server

handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
log = logging.getLogger('kademlia')
log.addHandler(handler)
log.setLevel(logging.DEBUG)


loop = asyncio.get_event_loop()
loop.set_debug(True)

server = Server()
loop.run_until_complete(server.listen(8468))

try:
    loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    server.stop()
    loop.close()

Node2代码（set操作）

import logging
import asyncio
import sys

from kademlia.network import Server

if len(sys.argv) != 5:
    print("Usage: python set.py <bootstrap node> <bootstrap port> <key> <value>")
    sys.exit(1)

handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
log = logging.getLogger('kademlia')
log.addHandler(handler)
log.setLevel(logging.DEBUG)

async def run():
    server = Server()
    await server.listen(8469)
    bootstrap_node = (sys.argv[1], int(sys.argv[2]))
    await server.bootstrap([bootstrap_node])
    await server.set(sys.argv[3], sys.argv[4])

    # Query the DHT for the key on node2
    retrieved_value = await server.get(sys.argv[3])
    print(f"Node 2 retrieved value from DHT: {retrieved_value}")
    #server.stop()

asyncio.run(run())

Node3代码（get操作）

import logging
import asyncio
import sys

from kademlia.network import Server

if len(sys.argv) != 4:
    print("Usage: python get.py <bootstrap node> <bootstrap port> <key>")
    sys.exit(1)

handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
log = logging.getLogger('kademlia')
log.addHandler(handler)
log.setLevel(logging.DEBUG)

async def run():
    server = Server()
    await server.listen(8470)
    bootstrap_node = (sys.argv[1], int(sys.argv[2]))
    await server.bootstrap([bootstrap_node])

    result = await server.get(sys.argv[3])
    print("Get result:", result)
    server.stop()

asyncio.run(run())

具体实现方案 #

1. 检测离线节点 #

通过主动ping路由表中的节点判断是否在线，或利用lonely_buckets()定位长期无活跃节点的桶（这类桶内节点大概率离线）。实现定期检查的异步函数：

async def check_offline_nodes(server):
    # 遍历路由表所有桶中的节点
    for bucket in server.protocol.router.buckets:
        for contact in bucket.contacts.copy():  # 用copy避免遍历中修改列表
            try:
                # 发送ping请求，超时2秒则判定离线
                await server.protocol.ping(contact, timeout=2)
            except asyncio.TimeoutError:
                print(f"节点 {contact} 已离线，开始移除")
                await remove_and_refresh(server, contact)

2. 移除无效节点并刷新路由表 #

组合使用remove_contact()、flush()及主动节点查找完成路由表维护：

async def remove_and_refresh(server, offline_node):
    # 从路由表中移除离线节点
    server.protocol.router.remove_contact(offline_node)
    print(f"已成功移除离线节点 {offline_node}")
    
    # 刷新路由表持久化状态（避免重启后重新加载无效节点）
    server.protocol.router.flush()
    
    # 针对空桶补充新节点：遍历所有"孤独桶"，发起节点查找请求
    for bucket_id in server.protocol.router.lonely_buckets():
        # 生成当前桶范围内的随机ID，触发find_node操作发现新节点
        target_id = server.protocol.router._random_id_in_bucket(bucket_id)
        await server.protocol.find_node(target_id)

3. 整合到现有节点代码 #

以Node2为例，修改run()函数，添加定期检查的后台任务，保持节点持续运行：

async def run():
    server = Server()
    await server.listen(8469)
    bootstrap_node = (sys.argv[1], int(sys.argv[2]))
    await server.bootstrap([bootstrap_node])
    await server.set(sys.argv[3], sys.argv[4])

    # Query the DHT for the key on node2
    retrieved_value = await server.get(sys.argv[3])
    print(f"Node 2 retrieved value from DHT: {retrieved_value}")
    
    # 定义定期检查任务，每30秒执行一次
    async def periodic_check():
        while True:
            await check_offline_nodes(server)
            await asyncio.sleep(30)
    
    # 启动后台检查任务
    asyncio.create_task(periodic_check())
    
    # 保持节点运行，直到收到中断信号
    try:
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        server.stop()

asyncio.run(run())

初始节点也可添加相同的定期检查逻辑，确保路由表实时维护。

关于刷新操作的必要性 #

• remove_contact()仅修改路由表的内存结构，flush()会将当前路由表状态写入持久化存储（默认是本地文件），每次移除节点后建议调用flush()，避免节点重启后重新加载无效节点。
• 主动调用find_node补充空桶的操作并非强制，但能快速恢复路由表的健壮性；若不主动执行，Kademlia在后续的get/set操作中也会自动发现新节点，但主动刷新能缩短恢复周期。

内容的提问来源于stack exchange，提问作者user23347973