第6章：隊列與處理器
在第5章：分類器中，我們了解了系統如何分析原始種子數據。但當系統突然發現數百萬新種子時，如何高效處理這些海量任務？這就是隊列與處理器系統的職責所在。
核心概念
任務隊列
- 功能定位：如同工廠的傳送帶，有序管理所有待處理任務
- 核心特性：
- 自動重試機制（失敗任務最多重試2次）
- 優先級排序（高優先級任務優先執行）
- 任務去重（通過指紋哈希防止重復）
處理器
- 工作模式：從隊列獲取任務并執行具體操作
- 并發控制：每個隊列可配置獨立的工作線程數
- 超時機制：默認單任務最長執行時間30分鐘
任務生命周期
狀態流轉
數據庫結構
type QueueJob struct {ID string // 任務唯一標識Queue string // 所屬隊列名(如"process_torrent")Status string // 任務狀態(pending/running/completed)Payload string // 任務參數(JSON格式)Retries uint // 當前重試次數MaxRetries uint // 最大重試次數(默認2)Priority int // 優先級(數值越大優先級越高)
}
實戰應用
批量重新分類
通過命令行觸發電影類種子重新分類：
bitmagnet worker reprocess-torrents \--content-type movie \--classify-mode rematch
自定義工作流
- 創建處理任務：
msg := processor.MessageParams{InfoHashes: []protocol.ID{hash1, hash2},ClassifyMode: processor.ClassifyModeRematch,
}
job, _ := model.NewQueueJob("process_torrent", msg)
- 提交任務隊列：
db.Create(&job) // 任務進入pending狀態
技術實現
處理器邏輯
func (p processor) Process(ctx context.Context, params MessageParams) error {// 1. 從數據庫加載種子數據torrents, _ := p.search.TorrentsWithMissingInfoHashes(ctx, params.InfoHashes)// 2. 調用分類器處理for _, torrent := range torrents {result, _ := p.classifier.Run(ctx, torrent)// 3. 保存分類結果p.dao.TorrentContent.Create(&model.TorrentContent{InfoHash: torrent.InfoHash,ContentType: result.ContentType,})}return nil
}
隊列服務
func (s server) runWorker(ctx context.Context, h handler.Handler) {for {// 1. 獲取待處理任務job, _ := s.query.QueueJob.Where(q.Queue.Eq(h.Queue),q.Status.Eq("pending"),).First()// 2. 標記任務為執行中s.query.QueueJob.Where(q.ID.Eq(job.ID)).Update("status", "running")// 3. 執行處理器邏輯if err := h.Handle(ctx, job); err != nil {// 處理失敗邏輯} else {// 標記任務完成}}
}
總結
隊列與處理器系統通過：
- 異步任務管理
- 自動容錯機制
- 優先級調度
保障系統穩定處理海量任務。下一章將深入DHT網絡核心組件：DHT路由表
第7章：DHT路由表
在第6章：隊列與處理器中，我們了解了系統如何管理后臺任務。本章將深入探索DHT爬蟲的核心導航系統——DHT路由表。
路由表解析
核心功能
路由表如同智能地址簿，實現：
- 節點管理：記錄已知BitTorrent客戶端（節點）的ID與網絡地址
- 哈希索引：存儲種子哈希值與對應節點關系
- 智能檢索：基于ID相似度快速定位最近節點
- 動態更新：持續淘汰失效節點（默認超時30分鐘）
關鍵參數
| 參數名 | 默認值 | 說明 |
|---|---|---|
| nodesK | 80 | 單節點桶最大容量 |
| hashesK | 80 | 單哈希桶最大容量 |
| nodeTimeout | 30m | 節點無響應淘汰閾值 |
數據結構
節點結構
type Node struct {ID [20]byte // 節點唯一標識Addr netip.AddrPort // IP地址與端口LastRespondedAt time.Time // 最后響應時間IsCandidate bool // 是否適合采樣請求
}
哈希記錄
type Hash struct {ID [20]byte // 種子哈希值Peers []Peer // 已知持有節點AddedAt time.Time // 發現時間
}type Peer struct {Addr netip.AddrPort // 節點網絡地址
}
核心操作
節點管理
哈希檢索
func (t *Table) GetClosestHashes(targetID [20]byte, limit int) []Hash {return t.btree.Closest(targetID, limit)
}
監控指標
通過Prometheus暴露的關鍵指標：
- bitmagnet_dht_ktable_nodes_count：當前活躍節點數
- bitmagnet_dht_ktable_hashes_added_total：累計發現哈希數
- bitmagnet_dht_ktable_nodes_dropped_total：淘汰節點計數
實現原理
接口定義
type Table interface {PutNode(ID, netip.AddrPort) error // 添加節點DropNode(ID, error) bool // 移除節點GetClosestNodes(ID, int) []Node // 獲取最近節點PutHash(ID, []Peer) error // 記錄哈希
}
B樹索引
type Btree struct {root *bucketsize intmutex sync.RWMutex
}func (b *Btree) Closest(target [20]byte, n int) []ID {// 基于XOR距離算法查找最近鄰
}
總結
DHT路由表通過：
- 高效B樹索引
- 智能節點淘汰
- 實時監控體系
為爬蟲提供穩定的網絡導航能力。下一章將探索系統如何優化存儲結構：數據分片策略
