修改成redis

This commit is contained in:
2026-04-20 18:26:54 +08:00
parent e944a25e56
commit a9cb0b2481
17 changed files with 2408 additions and 933 deletions
+29 -53
View File
@@ -5,85 +5,61 @@ Python 原版的 Go 语言重构,使用标准英文命名,单二进制部署
## 目录结构
```
golang/
sese-engine-go/
├── main.go # 主入口,goroutine 启动所有模块
├── go.mod
├── config/
│ └── config.go # 全局配置参数(对应 配置.py
│ └── config.go # 全局配置参数
├── storage/
── storage.go # bbolt 持久化层(对应 存储.py,替换 rimo_storage
── types.go # 核心数据类型定义
│ └── store.go # Redis + MySQL 存储层
├── crawler/
│ ├── crawler.go # BFS 爬虫调度(对应 上网.py
│ └── fetcher.go # HTTP 获取 + robots.txt + 限流(对应 虫.py
│ ├── crawler.go # BFS 爬虫调度
│ └── fetcher.go # HTTP 获取 + robots.txt + 限流
├── parser/
│ └── parser.go # HTML 解析(对应 文.py
│ └── parser.go # HTML 解析
├── analyzer/
│ └── analyzer.go # 分词 + 关键词权重(对应 分析.py + utils.py 分词部分)
│ 使用 gojieba(中文)+ gofasttext(语言检测)
├── harvester/
│ └── harvester.go # 索引写入服务,监听 :5000(对应 收获服务器.py)
│ └── analyzer.go # 分词 + 关键词权重 + 语言检测
├── search/
│ └── server.go # 搜索 API,监听 :80(对应 人服务器.py
│ └── server.go # 搜索 API + 收获服务
├── backlink/
│ └── backlink.go # 反向链接计算,每 48h 运行(对应 回.py
│ └── backlink.go # 反向链接计算
└── info/
└── info.go # 繁荣表 / 调整表 / 屏蔽词加载(对应 信息.py
└── info.go # 繁荣表 / 调整表 / 屏蔽词
```
## 依赖项
| Go 包 | 替代 Python 包 | 用途 |
|-------|--------------|------|
| `github.com/yanyiwu/gojieba` | `jieba` | 中文分词 |
| `github.com/nicholasgasior/gofasttext` | `fasttext` | 语言检测 |
| `go.etcd.io/bbolt` | `rimo_storage` | KV 存储 / 倒排索引 |
| `github.com/andybalholm/brotli` | `brotli` | 压缩 |
| `golang.org/x/net/html` | `lxml` | HTML 解析 |
| `golang.org/x/net/html/charset` | chardet | 编码检测 |
| Go 包 | 用途 |
|-------|------|
| `github.com/yanyiwu/gojieba` | 中文分词 |
| `github.com/pemistahl/lingua-go` | 语言检测(内置模型) |
| `github.com/redis/go-redis/v9` | Redis 客户端(高性能内存存储) |
| `github.com/go-sql-driver/mysql` | MySQL 驱动(持久化存储) |
| `github.com/andybalholm/brotli` | Brotli 压缩 |
| `golang.org/x/net/html` | HTML 解析 |
## 构建与运行
```bash
cd golang
# 下载依赖(需要 CGo 编译器,用于 gojieba / gofasttext
# 下载依赖(需要 CGo 编译器,用于 gojieba
go mod tidy
# 构建
go build -o sese-engine .
# 运行(在 sese-engine 项目根目录下)
cd ..
./golang/sese-engine \
--storage ./savedata \
--entry https://zh.wikipedia.org/ \
--fasttext ./lid.176.ftz \
--stopwords ./data/标点符号.json
# 运行
./sese-engine --storage ./savedata --entry https://zh.wikipedia.org/
```
一个进程启动所有模块:
- `:5000` — 收获服务器(爬虫推送关键词)
- `:80` — 搜索 API`GET /search?q=关键词`
- 后台 goroutine — BFS 爬虫
- 后台 goroutine — 每 48 小时反向链接计算
## 架构说明
## 与 Python 版的主要差异
| 方面 | Python 版 | Go 版 |
|------|---------|-------|
| 并发 | GIL + 线程池(假并发) | goroutine 真并发 |
| 存储 | rimo_storage(自研)| bbolt(嵌入式 KV |
| 部署 | 需要 Python 环境 | 单二进制,无运行时依赖 |
| 命名 | 全中文 | 标准英文 |
| 进程数 | 3~4 个进程 | 1 个进程多 goroutine |
| 编码检测 | requests 自动检测 | `golang.org/x/net/html/charset` |
| Prometheus | 可选 | 暂未集成(可后续添加) |
- **Redis**:高性能内存存储,用于实时索引读写
- **MySQL**:持久化存储,用于数据备份和复杂查询
- **内存聚合**:写入先到 Redis,定期合并到 MySQL
## 注意事项
1. **CGo 依赖**gojieba 和 gofasttext 均需要 C/C++ 编译器(gcc/clang)。
Windows 下建议使用 MinGW 或 WSL
2. **fasttext 模型**`lid.176.ftz` 需要与 Python 版共用,路径通过 `--fasttext` 指定
3. **数据迁移**:存储格式(bbolt JSON)与 Python 版(rimo_storage 二进制)不兼容,
需要全新爬取,或编写迁移脚本。
4. **stop words 文件**:复用 Python 版的 `data/标点符号.json`
1. **CGo 依赖**gojieba 需要 C/C++ 编译器(gcc/clang)。
2. **Redis + MySQL**:需要提前部署并配置连接参数
3. **配置文件**`config/config.yml` 中配置数据库连接信息
+6 -6
View File
@@ -21,7 +21,7 @@ import (
// Runner 管理反向链接计算循环。
type Runner struct {
db *storage.DB
store *storage.RedisStoreV2
storagePath string // 存储根目录(用于写入 prosper.json
mu sync.Mutex
running bool // 是否正在计算中
@@ -31,8 +31,8 @@ type Runner struct {
}
// New 创建一个 Runner 实例。
func New(db *storage.DB, storagePath string) *Runner {
return &Runner{db: db, storagePath: storagePath}
func New(store *storage.RedisStoreV2, storagePath string) *Runner {
return &Runner{store: store, storagePath: storagePath}
}
// Status 返回反链计算器的当前状态快照。
@@ -179,7 +179,7 @@ func (r *Runner) collectStats() *siteStats {
serverCount: make(map[string]int),
}
_ = r.db.ForEachSite(func(host string, info *storage.SiteInfo) error {
_ = r.store.ForEachSite(func(host string, info *storage.SiteInfo) error {
super := superDomain(host)
stats.subdomainCount[super]++
@@ -235,7 +235,7 @@ func (r *Runner) aggregate(filter func(*storage.SiteInfo) bool, stats *siteStats
pruneThreshold := 0.02
i := 0
_ = r.db.ForEachSite(func(host string, info *storage.SiteInfo) error {
_ = r.store.ForEachSite(func(host string, info *storage.SiteInfo) error {
if filter != nil && !filter(info) {
return nil
}
@@ -336,7 +336,7 @@ func (r *Runner) aggregateWithScores(scores map[string]float64, stats *siteStats
serverTable := buildServerTable(stats.serverCount)
vectors := make(map[string][]float32)
_ = r.db.ForEachSite(func(host string, info *storage.SiteInfo) error {
_ = r.store.ForEachSite(func(host string, info *storage.SiteInfo) error {
score, ok := scores[host]
if !ok || strings.Contains(host, "/") {
return nil
+115
View File
@@ -0,0 +1,115 @@
# ============================================
# sese-engine 配置文件示例
# ============================================
index:
max_urls_per_key: 11000 # 每个关键词最大URL数量
max_same_domain_per_key: 20 # 每个关键词同域名最大数量
big_clean_threshold: 2000000 # 大清理阈值
max_new_urls_per_key: 10000 # 每个关键词最大新增URL
min_urls_for_new_key: 3 # 新关键词最小URL数量
crawler:
spider_name: "Haibara_AI_spider" # 爬虫名称
cooldown: 3 # 请求间隔(秒)
workers: 22 # 并发数
crawl_focus: 0.7 # 爬取聚焦系数
max_keywords_per_page: 250 # 每页最大关键词数
max_epoch: 100 # 最大轮数
expected_prosper_ratio: 0.6 # 期望繁荣值比例
entry_url: "https://haibara.ai/" # 入口URL
max_page_size: 5242880 # 单页最大5MB
recrawl_max_age: 2592000 # 重爬过期30天
recrawl_check_interval: 3600 # 检查间隔1小时
recrawl_batch_size: 500 # 每批500个
max_priority_children: 100 # 优先队列子链接上限
search:
use_online_snippet: true
online_snippet_timeout: 3
weight_daily_decay: 0.996
language_weight: 0.5
consecutive_key_weight: 1.3
backlink_weight: 1.0
server_port: 50082
flush_interval_seconds: 300
stats_refresh_interval: 30
miss_penalty: 0.15
unix_socket: "" # Unix Socket路径(可选)
backlink:
baseline: 200000 # 反链基准值
storage:
path: "./savedata" # 存储路径
# ============================================
# MySQL 配置(默认关闭,启用时需设置 enabled: true
# ============================================
# 支持两种连接方式:Unix Socket 和 TCP
# 优先级:UnixSocket > TCP(如果UnixSocket非空则优先使用)
mysql:
# 是否启用 MySQL(默认 false,不启用则不连接、不刷盘)
enabled: false
# 连接方式(可选,默认tcp,可设为 "socket" 或 "tcp"
# network: "tcp"
# ----- Unix Socket 连接(推荐,本地MySQL性能更好)-----
# 设置为 MySQL socket 路径即可,TCP配置将被忽略
unix_socket: "/var/run/mysqld/mysqld.sock"
# unix_socket: "/tmp/mysql.sock" # macOS 常见路径
# unix_socket: "" # 留空则使用TCP连接
# ----- TCP 连接 -----
host: "localhost"
port: 3306
# ----- 认证信息 -----
user: "root"
password: "your_password_here"
database: "sese_engine"
# ----- 连接池配置 -----
conn_max_lifetime: 3600 # 连接最大生命周期(秒),默认1小时
max_idle_conns: 10 # 最大空闲连接数
max_open_conns: 100 # 最大打开连接数
# ============================================
# Redis 配置
# ============================================
# 支持两种连接方式:Unix Socket 和 TCP
# 优先级:UnixSocket > TCP(如果UnixSocket非空则优先使用)
redis:
# 连接方式(可选,默认tcp
# network: "tcp"
# ----- Unix Socket 连接(推荐,本地Redis性能更好)-----
# 设置为 Redis socket 路径即可,TCP配置将被忽略
unix_socket: "/var/run/redis/redis.sock"
# unix_socket: "/tmp/redis.sock" # macOS 常见路径
# unix_socket: "" # 留空则使用TCP连接
# ----- TCP 连接 -----
host: "localhost"
port: 6379
# ----- 认证信息(无密码则留空)-----
password: ""
# ----- 数据库配置 -----
# 数据库编号(0-15),默认使用15号数据库
db: 15
# ----- 连接池配置 -----
pool_size: 100 # 连接池最大连接数
min_idle_conns: 10 # 最小空闲连接数
# ----- 超时配置(毫秒)-----
read_timeout: 500 # 读超时
write_timeout: 500 # 写超时
prometheus:
crawler_port: 14950
backlink_port: 14952
search_port: 14953
+193
View File
@@ -23,6 +23,8 @@ type Config struct {
Search SearchConfig `yaml:"search"`
Backlink BacklinkConfig `yaml:"backlink"`
Storage StorageConfig `yaml:"storage"`
MySQL MySQLConfig `yaml:"mysql"`
Redis RedisConfig `yaml:"redis"`
Prometheus PrometheusConfig `yaml:"prometheus"`
}
@@ -77,6 +79,62 @@ type StorageConfig struct {
Path string `yaml:"path"`
}
// MySQLConfig MySQL数据库连接配置
// 支持两种连接方式:Unix Socket 和 TCP
// 优先级:UnixSocket > TCP(如果UnixSocket非空则优先使用)
type MySQLConfig struct {
// 是否启用 MySQL(默认关闭)
Enabled bool `yaml:"enabled"`
// 连接方式: "socket" 或 "tcp"(自动推断,可不填)
Network string `yaml:"network"`
// Unix Socket 路径(Linux/macOS),优先使用
// 示例: "/var/run/mysqld/mysqld.sock" 或 "/tmp/mysql.sock"
UnixSocket string `yaml:"unix_socket"`
// TCP 连接方式:服务器地址
Host string `yaml:"host"`
// TCP 连接方式:端口号
Port int `yaml:"port"`
// 用户名
User string `yaml:"user"`
// 密码
Password string `yaml:"password"`
// 数据库名
Database string `yaml:"database"`
// 连接超时时间(秒)
ConnMaxLifetime int `yaml:"conn_max_lifetime"` // 秒
// 最大空闲连接数
MaxIdleConns int `yaml:"max_idle_conns"`
// 最大打开连接数
MaxOpenConns int `yaml:"max_open_conns"`
}
// RedisConfig Redis连接配置
// 支持两种连接方式:Unix Socket 和 TCP
// 优先级:UnixSocket > TCP(如果UnixSocket非空则优先使用)
type RedisConfig struct {
// 连接方式: "socket" 或 "tcp"(自动推断,可不填)
Network string `yaml:"network"`
// Unix Socket 路径,优先使用
// 示例: "/var/run/redis/redis.sock" 或 "/tmp/redis.sock"
UnixSocket string `yaml:"unix_socket"`
// TCP 连接方式:服务器地址
Host string `yaml:"host"`
// TCP 连接方式:端口号
Port int `yaml:"port"`
// 密码(无密码则留空)
Password string `yaml:"password"`
// 数据库编号(0-15),默认 15
DB int `yaml:"db"`
// 池大小(最大连接数)
PoolSize int `yaml:"pool_size"`
// 最小空闲连接数
MinIdleConns int `yaml:"min_idle_conns"`
// 读超时时间(毫秒)
ReadTimeout int `yaml:"read_timeout"` // 毫秒
// 写超时时间(毫秒)
WriteTimeout int `yaml:"write_timeout"` // 毫秒
}
// PrometheusConfig Prometheus监控端口配置
type PrometheusConfig struct {
CrawlerPort int `yaml:"crawler_port"`
@@ -217,6 +275,31 @@ func GetDefaultConfig() Config {
Storage: StorageConfig{
Path: "./savedata",
},
MySQL: MySQLConfig{
Enabled: false,
Network: "tcp",
UnixSocket: "",
Host: "localhost",
Port: 3306,
User: "root",
Password: "",
Database: "sese_engine",
ConnMaxLifetime: 3600, // 1小时
MaxIdleConns: 10,
MaxOpenConns: 100,
},
Redis: RedisConfig{
Network: "tcp",
UnixSocket: "",
Host: "localhost",
Port: 6379,
Password: "",
DB: 15, // 默认使用15号数据库
PoolSize: 100,
MinIdleConns: 10,
ReadTimeout: 500, // 毫秒
WriteTimeout: 500, // 毫秒
},
Prometheus: PrometheusConfig{
CrawlerPort: 14950,
BacklinkPort: 14952,
@@ -357,3 +440,113 @@ func MaxPriorityChildren() int {
// 为了向后兼容,保留 StoragePath 常量
const StoragePath = "./savedata"
// ---- MySQL 配置访问函数 ----
// MySQLEnabled 返回是否启用 MySQL(默认关闭)
func MySQLEnabled() bool {
return Global.MySQL.Enabled
}
// MySQLDSN 返回 MySQL 连接字符串(DSN)
// 根据配置自动选择 Unix Socket 或 TCP 方式
func MySQLDSN() string {
cfg := Global.MySQL
if cfg.UnixSocket != "" {
// 使用 Unix Socket 连接(推荐,本地连接性能更好)
return cfg.User + ":" + cfg.Password + "@unix(" + cfg.UnixSocket + ")/" + cfg.Database + "?parseTime=true&loc=Local"
}
// 使用 TCP 连接
return cfg.User + ":" + cfg.Password + "@tcp(" + cfg.Host + ":" + itoa(cfg.Port) + ")/" + cfg.Database + "?parseTime=true&loc=Local"
}
// MySQLConnMaxLifetime 返回连接最大生命周期(秒)
func MySQLConnMaxLifetime() int {
if Global.MySQL.ConnMaxLifetime <= 0 {
return 3600
}
return Global.MySQL.ConnMaxLifetime
}
// MySQLMaxIdleConns 返回最大空闲连接数
func MySQLMaxIdleConns() int {
if Global.MySQL.MaxIdleConns <= 0 {
return 10
}
return Global.MySQL.MaxIdleConns
}
// MySQLMaxOpenConns 返回最大打开连接数
func MySQLMaxOpenConns() int {
if Global.MySQL.MaxOpenConns <= 0 {
return 100
}
return Global.MySQL.MaxOpenConns
}
// ---- Redis 配置访问函数 ----
// RedisAddr 返回 Redis 连接地址
// 根据配置自动选择 Unix Socket 或 TCP 方式
func RedisAddr() string {
cfg := Global.Redis
if cfg.UnixSocket != "" {
return cfg.UnixSocket
}
return cfg.Host + ":" + itoa(cfg.Port)
}
// RedisPoolSize 返回连接池大小
func RedisPoolSize() int {
if Global.Redis.PoolSize <= 0 {
return 100
}
return Global.Redis.PoolSize
}
// RedisMinIdleConns 返回最小空闲连接数
func RedisMinIdleConns() int {
if Global.Redis.MinIdleConns <= 0 {
return 10
}
return Global.Redis.MinIdleConns
}
// RedisDB 返回数据库编号
func RedisDB() int {
return Global.Redis.DB
}
// RedisPassword 返回密码(空字符串表示无密码)
func RedisPassword() string {
return Global.Redis.Password
}
// RedisReadTimeout 返回读超时时间(毫秒)
func RedisReadTimeout() int {
if Global.Redis.ReadTimeout <= 0 {
return 500
}
return Global.Redis.ReadTimeout
}
// RedisWriteTimeout 返回写超时时间(毫秒)
func RedisWriteTimeout() int {
if Global.Redis.WriteTimeout <= 0 {
return 500
}
return Global.Redis.WriteTimeout
}
// itoa 将 int 转换为字符串(避免导入 strconv)
func itoa(n int) string {
if n == 0 {
return "0"
}
result := ""
for n > 0 {
result = string(rune('0'+n%10)) + result
n /= 10
}
return result
}
+15 -15
View File
@@ -52,7 +52,7 @@ const (
// Crawler 编排整个 BFS 爬取流程。
type Crawler struct {
fetcher *Fetcher // HTTP 抓取器(含 robots.txt 和限流)
db *storage.DB // 持久化数据库
store *storage.RedisStoreV2 // 持久化存储
analyzer *analyzer.Analyzer // 分词和关键词分析
prosperMap map[string]float64 // 域名 → 反向链接繁荣值(来自 info 模块,越大越"有价值")
stats Stats // 原子计数器
@@ -164,10 +164,10 @@ func DecrementPriorityLevel2Inflight(n int64) {
// New 创建一个 Crawler 实例。
// prosperMap 由 info 模块加载,传入域名繁荣值用于调度优先级计算。
func New(db *storage.DB, a *analyzer.Analyzer, prosperMap map[string]float64) *Crawler {
func New(store *storage.RedisStoreV2, a *analyzer.Analyzer, prosperMap map[string]float64) *Crawler {
c := &Crawler{
fetcher: NewFetcher(config.SpiderName(), time.Duration(config.CrawlerCooldown())*time.Second),
db: db,
store: store,
analyzer: a,
prosperMap: prosperMap,
visited: make(map[string]bool),
@@ -192,7 +192,7 @@ func (c *Crawler) warmVisited() {
expired := 0
maxAge := int64(config.RecrawlMaxAge())
now := time.Now().Unix()
_ = c.db.ForEachSnippet(func(u string, entry *storage.SnippetEntry) error {
_ = c.store.ForEachSnippet(func(u string, entry *storage.SnippetEntry) error {
if now-entry.Timestamp < maxAge {
c.visited[u] = true // 未过期,仍然跳过
count++
@@ -221,7 +221,7 @@ func (c *Crawler) startRecrawlTicker() {
removed := 0
c.visitedMu.Lock()
_ = c.db.ForEachSnippet(func(u string, entry *storage.SnippetEntry) error {
_ = c.store.ForEachSnippet(func(u string, entry *storage.SnippetEntry) error {
if removed >= batchSize {
return fmt.Errorf("batch full") // 提前终止遍历
}
@@ -365,7 +365,7 @@ func (c *Crawler) priorityCrawlLoop(rawURL string, level int) {
}()
// 标记 DB 中该 URL 为已访问,防止重启后再次被调度
_ = c.db.MarkPriorityURLVisited(rawURL)
_ = c.store.MarkPriorityURLVisited(rawURL)
// 两级都不限制子链接数量
children := c.visitURLUnlimited(rawURL)
@@ -441,7 +441,7 @@ func (c *Crawler) isVisited(url string) bool {
// 将未访问的插入队列前端(prepend),已爬取的条目从存储中清除。
// 返回本次插入队列的 URL 数量。
func (c *Crawler) fetchAndApplyPriorityURLs(queue *[]string) int {
entries, err := c.db.GetPriorityURLs()
entries, err := c.store.GetPriorityURLs()
if err != nil || len(entries) == 0 {
return 0
}
@@ -449,14 +449,14 @@ func (c *Crawler) fetchAndApplyPriorityURLs(queue *[]string) int {
added := 0
for _, e := range entries {
if c.isVisited(e.URL) {
_ = c.db.RemovePriorityURL(e.URL)
_ = c.store.RemovePriorityURL(e.URL)
continue
}
*queue = append([]string{e.URL}, *queue...)
added++
}
_ = c.db.ClearVisitedPriorityURLs()
_ = c.store.ClearVisitedPriorityURLs()
return added
}
@@ -713,7 +713,7 @@ func (c *Crawler) visitURLRaw(rawURL string, forceIndex bool) (title, desc, text
// 增量重爬检测:查询上次爬取的哈希,内容未变则跳过关键词提取
isRecrawl := false
oldEntry, _ := c.db.GetSnippet(res.FinalURL)
oldEntry, _ := c.store.GetSnippet(res.FinalURL)
if !forceIndex && oldEntry != nil && oldEntry.ContentHash != "" && oldEntry.ContentHash == contentHash {
isRecrawl = true
//log.Printf("[crawler] unchanged (recrawl skip): %s", res.FinalURL)
@@ -721,7 +721,7 @@ func (c *Crawler) visitURLRaw(rawURL string, forceIndex bool) (title, desc, text
// 缓存 URL 摘要(仅对短 URL 缓存,防止超长 URL 浪费空间)
if len(res.FinalURL) < 250 {
_ = c.db.SetSnippet(res.FinalURL, &storage.SnippetEntry{
_ = c.store.SetSnippet(res.FinalURL, &storage.SnippetEntry{
Title: title,
Description: truncate(desc, 256),
Text: truncate(text, 256),
@@ -756,7 +756,7 @@ func (c *Crawler) visitURLRaw(rawURL string, forceIndex bool) (title, desc, text
if fromHost == "" {
continue
}
_ = c.db.UpdateSiteInfo(fromHost, func(info *storage.SiteInfo) {
_ = c.store.UpdateSiteInfo(fromHost, func(info *storage.SiteInfo) {
if info.Redirects == nil {
info.Redirects = make(map[string]string)
}
@@ -799,7 +799,7 @@ func (c *Crawler) updateSiteFailure(rawURL string) {
if host == "" {
return
}
_ = c.db.UpdateSiteInfo(host, func(info *storage.SiteInfo) {
_ = c.store.UpdateSiteInfo(host, func(info *storage.SiteInfo) {
if info.SuccessRate == nil {
zero := 0.0
info.SuccessRate = &zero
@@ -831,7 +831,7 @@ func (c *Crawler) updateSiteSuccess(host string, res *FetchResult, title, desc,
}
sampled := sampleStrings(external, 10)
_ = c.db.UpdateSiteInfo(host, func(info *storage.SiteInfo) {
_ = c.store.UpdateSiteInfo(host, func(info *storage.SiteInfo) {
// 访问计数 +1,更新最后访问时间
info.VisitCount++
info.LastVisitTime = now
@@ -978,7 +978,7 @@ func (c *Crawler) schedule(links []URLWeight) []string {
wg.Add(1)
go func(host string) {
defer wg.Done()
info, _ := c.db.GetSiteInfo(host)
info, _ := c.store.GetSiteInfo(host)
mu.Lock()
siteCache[host] = info
mu.Unlock()
+8 -1
View File
@@ -14,6 +14,7 @@ import (
"time" // 时间(限流间隔计算、robots.txt 缓存过期)
"golang.org/x/net/html/charset" // HTML 字符集自动检测(将各种编码转为 UTF-8)
"golang.org/x/text/encoding/simplifiedchinese" // GBK → UTF-8 转换兜底
)
// ErrCrawl 表示爬取过程中的预期错误(404、被 robots.txt 禁止、非 HTML 类型等)。
@@ -341,11 +342,17 @@ func decodeBody(r io.Reader, contentType string, sizeLimit int) (string, error)
// 使用 golang.org/x/net/html/charset 自动检测 HTML 编码并转为 UTF-8
utf8Reader, err := charset.NewReader(reader, contentType)
if err != nil {
// 备选方案:直接以 UTF-8 读取(可能乱码但不崩溃)
// charset 检测失败时,先读取原始字节,再尝试 GBK 兜底
data, readErr := io.ReadAll(reader)
if readErr != nil {
return "", readErr
}
// 将 GBK 字节流转为 UTF-8 字符串
utf8Bytes, convErr := simplifiedchinese.GBK.NewDecoder().Bytes(data)
if convErr == nil {
return string(utf8Bytes), nil
}
// 转换失败则返回原始字节(可能乱码但不崩溃)
return string(data), nil
}
data, err := io.ReadAll(utf8Reader)
+6 -4
View File
@@ -3,18 +3,20 @@ module sese-engine
go 1.21
require (
github.com/andybalholm/brotli v1.1.0
github.com/go-sql-driver/mysql v1.8.1
github.com/pemistahl/lingua-go v1.4.0
github.com/redis/go-redis/v9 v9.5.1
github.com/yanyiwu/gojieba v1.4.4
go.etcd.io/bbolt v1.3.9
golang.org/x/net v0.23.0
golang.org/x/text v0.14.0
gopkg.in/yaml.v3 v3.0.1
)
require (
filippo.io/edwards25519 v1.1.0 // indirect
github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/shopspring/decimal v1.3.1 // indirect
golang.org/x/exp v0.0.0-20221106115401-f9659909a136 // indirect
golang.org/x/sys v0.18.0 // indirect
golang.org/x/text v0.14.0 // indirect
google.golang.org/protobuf v1.31.0 // indirect
)
+14 -8
View File
@@ -1,7 +1,17 @@
github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M=
github.com/andybalholm/brotli v1.1.0/go.mod h1:sms7XGricyQI9K10gOSf56VKKWS4oLer58Q+mhRPtnY=
filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y=
github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
@@ -10,22 +20,18 @@ github.com/pemistahl/lingua-go v1.4.0 h1:ifYhthrlW7iO4icdubwlduYnmwU37V1sbNrwhKB
github.com/pemistahl/lingua-go v1.4.0/go.mod h1:ECuM1Hp/3hvyh7k8aWSqNCPlTxLemFZsRjocUf3KgME=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/redis/go-redis/v9 v9.5.1 h1:H1X4D3yHPaYrkL5X06Wh6xNVM/pX0Ft4RV0vMGvLBh8=
github.com/redis/go-redis/v9 v9.5.1/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5gKV8=
github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
github.com/yanyiwu/gojieba v1.4.4 h1:Iukkf8WlIfqAKtsGZjUhGR1ArKa7DtLDNmW8bvUI8JI=
github.com/yanyiwu/gojieba v1.4.4/go.mod h1:JUq4DddFVGdHXJHxxepxRmhrKlDpaBxR8O28v6fKYLY=
go.etcd.io/bbolt v1.3.9 h1:8x7aARPEXiXbHmtUwAIv7eV2fQFHrLLavdiJ3uzJXoI=
go.etcd.io/bbolt v1.3.9/go.mod h1:zaO32+Ti0PK1ivdPtgMESzuzL2VPoIG1PCQNvOdo/dE=
golang.org/x/exp v0.0.0-20221106115401-f9659909a136 h1:Fq7F/w7MAa1KJ5bt2aJ62ihqp9HDcRuyILskkpIAurw=
golang.org/x/exp v0.0.0-20221106115401-f9659909a136/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
golang.org/x/net v0.23.0 h1:7EYJ93RZ9vYSZAIb2x3lnuvqO5zneoD6IvWjuhfxjTs=
golang.org/x/net v0.23.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
golang.org/x/sync v0.5.0 h1:60k92dhOjHxJkrqnwsfl8KuaHbn/5dl0lUPUklKo3qE=
golang.org/x/sync v0.5.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+50 -11
View File
@@ -10,6 +10,7 @@
package main
import (
"context"
"flag" // 命令行参数解析
"fmt" // 格式化(搜索服务端口)
"log" // 日志输出
@@ -17,12 +18,15 @@ import (
"os/signal" // 信号捕获
"path/filepath" // 路径处理
"syscall" // 系统调用(SIGTERM
"time" // 时间(刷盘间隔)
"sese-engine/analyzer" // 文本分析和关键词提取
"sese-engine/backlink" // 反向链接(繁荣值)计算
"sese-engine/config" // 全局配置
"sese-engine/crawler" // BFS 爬虫
"sese-engine/info" // info 服务(繁荣表、调整表、屏蔽词)
"sese-engine/mysql" // MySQL 数据库连接
"sese-engine/redis" // Redis 连接
"sese-engine/search" // 搜索服务器(内嵌收获服务)
"sese-engine/storage" // 持久化存储
)
@@ -138,13 +142,44 @@ func main() {
log.SetFlags(log.LstdFlags | log.Lshortfile)
log.Printf("sese-engine starting storage=%s entry=%s", *storageDir, *entryURL)
// ---- 1. 存储层:打开 bbolt 数据库 ----
db, err := storage.Open(*storageDir)
if err != nil {
log.Fatalf("failed to open storage: %v", err)
// ---- 1. Redis 连接(高性能内存存储)----
if err := redis.Open(); err != nil {
log.Fatalf("failed to open redis: %v", err)
}
defer db.Close()
db.StartWriteFlusher() // 启动异步写缓冲后台刷盘
defer redis.Close()
// ---- 1.1 MySQL 连接(持久化存储,默认关闭)----
var flusher *mysql.Flusher
if config.MySQLEnabled() {
if err := mysql.Open(); err != nil {
log.Fatalf("failed to open mysql: %v", err)
}
defer mysql.Close()
// 从 MySQL 恢复数据到 Redis(如 Redis 数据丢失)
// 仅当 Redis 为空时才执行恢复,避免覆盖已有数据
ctx := context.Background()
size, _ := redis.Client.DBSize(ctx).Result()
if size > 0 {
log.Printf("[mysql-restore] Redis has %d keys, skipping restore", size)
} else if err := mysql.RestoreFromMySQLToRedis(redis.Client); err != nil {
log.Printf("[mysql-restore] warning: %v", err)
}
// MySQL 刷盘器:定期将 Redis 数据刷到 MySQL
flusher = mysql.NewFlusher(redis.Client, 5*time.Minute, 1000)
flusher.Start()
defer flusher.Stop()
} else {
log.Println("[mysql] disabled in config, skipping MySQL init")
}
// ---- 1.3 存储层:初始化 Redis 存储 ----
store := storage.NewRedisStoreV2()
if err := store.Init(); err != nil {
log.Fatalf("failed to init redis store: %v", err)
}
defer store.Close()
// ---- 2. Info 服务:加载繁荣表、调整表和屏蔽词 ----
infoSvc := info.New(*storageDir)
@@ -158,7 +193,7 @@ func main() {
defer anal.Close()
// ---- 4. 搜索服务器(默认 :80):对外提供搜索 API,同时内嵌收获服务(统一端口)
searchSrv := search.New(db, infoSvc, anal)
searchSrv := search.New(store, infoSvc, anal)
go func() {
addr := fmt.Sprintf(":%d", config.SearchServerPort())
if err := searchSrv.ListenAndServe(addr, config.UnixSocket()); err != nil {
@@ -167,14 +202,14 @@ func main() {
}()
// ---- 6. 反向链接计算器:每 48 小时运行一次 ----
bl := backlink.New(db, *storageDir)
bl := backlink.New(store, *storageDir)
searchSrv.SetBacklinkRunner(bl)
go bl.Run()
// ---- 7. 爬虫:从入口 URL 开始 BFS 爬取 ----
// 从 info 服务获取繁荣表快照,用于调度优先级决策
prosperMap := infoSvc.ProsperMap()
crawl := crawler.New(db, anal, prosperMap)
crawl := crawler.New(store, anal, prosperMap)
searchSrv.SetCrawler(crawl) // 注入爬虫用于 Priority URL 立即触发
go crawl.Run(*entryURL, config.MaxEpoch())
@@ -194,8 +229,12 @@ func main() {
crawl.WaitUntilStopped()
log.Println("crawler stopped")
// 最后刷盘,确保数据不丢失
log.Println("flushing index...")
// 最后刷盘Redis → Disk,条件刷 MySQL,确保数据不丢失
log.Println("flushing index to disk...")
searchSrv.Flush()
if flusher != nil {
log.Println("flushing data to mysql...")
flusher.RunAll() // 同步执行一次 MySQL 刷盘
}
log.Println("shutdown complete")
}
+445
View File
@@ -0,0 +1,445 @@
package mysql
import (
"context"
"fmt"
"log"
"strings"
"sync"
"time"
goredis "github.com/redis/go-redis/v9"
)
// Flusher 管理 Redis → MySQL 刷盘任务
type Flusher struct {
redisDB *goredis.Client // Redis 客户端引用
interval time.Duration // 刷盘间隔
batchSize int // 每批次处理数量
mu sync.Mutex // 防止并发刷盘
stopCh chan struct{}
wg sync.WaitGroup
}
// NewFlusher 创建刷盘器
func NewFlusher(redisDB *goredis.Client, interval time.Duration, batchSize int) *Flusher {
if interval <= 0 {
interval = 5 * time.Minute
}
if batchSize <= 0 {
batchSize = 1000
}
return &Flusher{
redisDB: redisDB,
interval: interval,
batchSize: batchSize,
stopCh: make(chan struct{}),
}
}
// Start 启动后台刷盘任务
func (f *Flusher) Start() {
f.wg.Add(1)
go func() {
defer f.wg.Done()
ticker := time.NewTicker(f.interval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
f.RunAll()
case <-f.stopCh:
log.Printf("[mysql-flusher] stopped")
return
}
}
}()
log.Printf("[mysql-flusher] started (interval=%v, batchSize=%d)", f.interval, f.batchSize)
}
// Stop 停止刷盘任务
func (f *Flusher) Stop() {
close(f.stopCh)
f.wg.Wait()
}
// RunAll 执行所有类型的刷盘
func (f *Flusher) RunAll() {
f.mu.Lock()
defer f.mu.Unlock()
start := time.Now()
log.Printf("[mysql-flusher] === starting flush ===")
// 刷盘顺序:snippet → site → index(按数据量从小到大)
f.flushSnippets()
f.flushSites()
f.flushIndex()
f.flushPriorityURLs()
log.Printf("[mysql-flusher] === flush done (took %v) ===", time.Since(start))
}
// flushSnippets 将 Redis gate:* 数据刷到 url_snippets 表
func (f *Flusher) flushSnippets() {
ctx := context.Background()
var cursor uint64
total := 0
for {
keys, nextCursor, err := f.redisDB.Scan(ctx, cursor, "gate:*", int64(f.batchSize)).Result()
if err != nil {
log.Printf("[mysql-flusher][snippets] scan error: %v", err)
return
}
if len(keys) > 0 {
f.batchUpsertSnippets(ctx, keys)
total += len(keys)
}
cursor = nextCursor
if cursor == 0 {
break
}
}
if total > 0 {
log.Printf("[mysql-flusher][snippets] flushed %d entries", total)
}
}
// batchUpsertSnippets 批量 upsert url_snippets
func (f *Flusher) batchUpsertSnippets(ctx context.Context, keys []string) {
if len(keys) == 0 || DB == nil {
return
}
query := `INSERT INTO url_snippets (url, url_hash, title, description, text, timestamp, content_hash)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
title = VALUES(title),
description = VALUES(description),
text = VALUES(text),
timestamp = VALUES(timestamp),
content_hash = VALUES(content_hash)`
tx, err := DB.BeginTx(ctx, nil)
if err != nil {
log.Printf("[mysql-flusher][snippets] begin tx error: %v", err)
return
}
defer tx.Rollback()
stmt, err := tx.PrepareContext(ctx, query)
if err != nil {
log.Printf("[mysql-flusher][snippets] prepare error: %v", err)
return
}
defer stmt.Close()
for _, key := range keys {
data, err := f.redisDB.HGetAll(ctx, key).Result()
if err != nil || len(data) == 0 {
continue
}
url := data["url"]
urlHash := data["url_hash"]
if urlHash == "" {
// 从 key 中提取 hashkey 格式:gate:<hash>
urlHash = strings.TrimPrefix(key, "gate:")
}
title := data["title"]
description := data["desc"]
text := data["text"]
ts := parseInt64(data["ts"])
contentHash := data["hash"]
_, err = stmt.ExecContext(ctx, url, urlHash, title, description, text, ts, contentHash)
if err != nil {
log.Printf("[mysql-flusher][snippets] exec error for %s: %v", url, err)
}
}
if err := tx.Commit(); err != nil {
log.Printf("[mysql-flusher][snippets] commit error: %v", err)
}
}
// flushSites 将 Redis site:* 数据刷到 site_info 表
func (f *Flusher) flushSites() {
ctx := context.Background()
var cursor uint64
total := 0
for {
keys, nextCursor, err := f.redisDB.Scan(ctx, cursor, "site:*", int64(f.batchSize)).Result()
if err != nil {
log.Printf("[mysql-flusher][sites] scan error: %v", err)
return
}
if len(keys) > 0 {
f.batchUpsertSites(ctx, keys)
total += len(keys)
}
cursor = nextCursor
if cursor == 0 {
break
}
}
if total > 0 {
log.Printf("[mysql-flusher][sites] flushed %d entries", total)
}
}
// batchUpsertSites 批量 upsert site_info
func (f *Flusher) batchUpsertSites(ctx context.Context, keys []string) {
if len(keys) == 0 || DB == nil {
return
}
query := `INSERT INTO site_info (host, visit_count, last_visit_time, fingerprint, success_rate,
html_structure, ips, quality, https_available, keywords, out_links,
languages, redirects, server_types)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
visit_count = VALUES(visit_count),
last_visit_time = VALUES(last_visit_time),
fingerprint = VALUES(fingerprint),
success_rate = VALUES(success_rate),
html_structure = VALUES(html_structure),
ips = VALUES(ips),
quality = VALUES(quality),
https_available = VALUES(https_available),
keywords = VALUES(keywords),
out_links = VALUES(out_links),
languages = VALUES(languages),
redirects = VALUES(redirects),
server_types = VALUES(server_types)`
tx, err := DB.BeginTx(ctx, nil)
if err != nil {
log.Printf("[mysql-flusher][sites] begin tx error: %v", err)
return
}
defer tx.Rollback()
stmt, err := tx.PrepareContext(ctx, query)
if err != nil {
log.Printf("[mysql-flusher][sites] prepare error: %v", err)
return
}
defer stmt.Close()
for _, key := range keys {
data, err := f.redisDB.HGetAll(ctx, key).Result()
if err != nil || len(data) == 0 {
continue
}
host := strings.TrimPrefix(key, "site:")
visitCount := int(parseInt64(data["visit_count"]))
lastVisitTime := parseInt64(data["last_visit_time"])
fingerprint := data["fingerprint"]
htmlStructure := data["html_structure"]
var successRate *float64
if v := data["success_rate"]; v != "" {
f := parseFloat(v)
successRate = &f
}
var quality *float64
if v := data["quality"]; v != "" {
q := parseFloat(v)
quality = &q
}
var httpsAvailable *int8
if v := data["https_available"]; v != "" {
i := int8(parseInt64(v))
httpsAvailable = &i
}
// JSON 字段:空字符串转为 NULL 或 "[]"
// MySQL JSON 类型不接受空字符串
ips := data["ips"]
if ips == "" {
ips = "[]"
}
keywords := data["keywords"]
if keywords == "" {
keywords = "[]"
}
outLinks := data["out_links"]
if outLinks == "" {
outLinks = "[]"
}
languages := data["languages"]
if languages == "" {
languages = "[]"
}
redirects := data["redirects"]
if redirects == "" {
redirects = "[]"
}
serverTypes := data["server_types"]
if serverTypes == "" {
serverTypes = "[]"
}
_, err = stmt.ExecContext(ctx, host, visitCount, lastVisitTime, fingerprint, successRate,
htmlStructure, ips, quality, httpsAvailable, keywords, outLinks,
languages, redirects, serverTypes)
if err != nil {
log.Printf("[mysql-flusher][sites] exec error for %s: %v", host, err)
}
}
if err := tx.Commit(); err != nil {
log.Printf("[mysql-flusher][sites] commit error: %v", err)
}
}
// flushIndex 将 Redis idx:* 数据刷到 index_entries 表
func (f *Flusher) flushIndex() {
ctx := context.Background()
var cursor uint64
total := 0
for {
keys, nextCursor, err := f.redisDB.Scan(ctx, cursor, "idx:*", int64(f.batchSize)).Result()
if err != nil {
log.Printf("[mysql-flusher][index] scan error: %v", err)
return
}
if len(keys) > 0 {
f.batchUpsertIndex(ctx, keys)
total += len(keys)
}
cursor = nextCursor
if cursor == 0 {
break
}
}
if total > 0 {
log.Printf("[mysql-flusher][index] flushed %d keywords", total)
}
}
// batchUpsertIndex 批量 upsert index_entries
func (f *Flusher) batchUpsertIndex(ctx context.Context, keys []string) {
if len(keys) == 0 || DB == nil {
return
}
tx, err := DB.BeginTx(ctx, nil)
if err != nil {
log.Printf("[mysql-flusher][index] begin tx error: %v", err)
return
}
defer tx.Rollback()
stmt, err := tx.PrepareContext(ctx, `INSERT INTO index_entries (keyword, url, weight)
VALUES (?, ?, ?)
ON DUPLICATE KEY UPDATE weight = VALUES(weight)`)
if err != nil {
log.Printf("[mysql-flusher][index] prepare error: %v", err)
return
}
defer stmt.Close()
for _, key := range keys {
keyword := strings.TrimPrefix(key, "idx:")
// 获取有序集合中的所有成员
entries, err := f.redisDB.ZRevRangeWithScores(ctx, key, 0, -1).Result()
if err != nil {
log.Printf("[mysql-flusher][index] zrange error for %s: %v", keyword, err)
continue
}
for _, entry := range entries {
url := entry.Member.(string)
weight := float32(entry.Score)
_, err = stmt.ExecContext(ctx, keyword, url, weight)
if err != nil {
log.Printf("[mysql-flusher][index] exec error for %s/%s: %v", keyword, url, err)
}
}
}
if err := tx.Commit(); err != nil {
log.Printf("[mysql-flusher][index] commit error: %v", err)
}
}
// flushPriorityURLs 将 Redis priority:* 数据刷到 priority_urls 表
func (f *Flusher) flushPriorityURLs() {
ctx := context.Background()
keys, err := f.redisDB.Keys(ctx, "priority:*").Result()
if err != nil {
log.Printf("[mysql-flusher][priority] keys error: %v", err)
return
}
if len(keys) == 0 || DB == nil {
return
}
tx, err := DB.BeginTx(ctx, nil)
if err != nil {
log.Printf("[mysql-flusher][priority] begin tx error: %v", err)
return
}
defer tx.Rollback()
stmt, err := tx.PrepareContext(ctx, `INSERT IGNORE INTO priority_urls (url) VALUES (?)`)
if err != nil {
log.Printf("[mysql-flusher][priority] prepare error: %v", err)
return
}
defer stmt.Close()
for _, key := range keys {
url := strings.TrimPrefix(key, "priority:")
_, err = stmt.ExecContext(ctx, url)
if err != nil {
log.Printf("[mysql-flusher][priority] exec error for %s: %v", url, err)
}
}
if err := tx.Commit(); err != nil {
log.Printf("[mysql-flusher][priority] commit error: %v", err)
return
}
log.Printf("[mysql-flusher][priority] flushed %d entries", len(keys))
}
// ============================================
// 辅助函数
// ============================================
func parseInt64(s string) int64 {
var v int64
fmt.Sscanf(s, "%d", &v)
return v
}
func parseFloat(s string) float64 {
var v float64
fmt.Sscanf(s, "%f", &v)
return v
}
+98
View File
@@ -0,0 +1,98 @@
-- ============================================
-- sese-engine MySQL 初始化脚本
-- 自动创建数据库(如果不存在)并创建表结构
-- ============================================
-- 创建数据库(如果不存在)
CREATE DATABASE IF NOT EXISTS sese_engine DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE DATABASE IF NOT EXISTS sese_test2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- ============================================
-- 倒排索引表
-- 存储关键词到URL的映射及其权重
-- ============================================
CREATE TABLE IF NOT EXISTS index_entries (
keyword VARCHAR(255) NOT NULL COMMENT '关键词',
url VARCHAR(2048) NOT NULL COMMENT 'URL地址',
weight FLOAT NOT NULL COMMENT '权重分数',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (keyword(255), url(255)),
INDEX idx_keyword (keyword),
INDEX idx_weight (weight),
INDEX idx_url (url(255))
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
COMMENT='倒排索引表';
-- ============================================
-- URL 摘要表
-- 存储每个URL的标题、描述、正文片段等
-- ============================================
CREATE TABLE IF NOT EXISTS url_snippets (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
url VARCHAR(2048) NOT NULL COMMENT 'URL地址(唯一)',
url_hash VARCHAR(64) NOT NULL COMMENT 'URL的MD5哈希(用于快速查询)',
title VARCHAR(512) COMMENT '页面标题',
description TEXT COMMENT 'meta description或自动生成的描述',
text MEDIUMTEXT COMMENT '正文文本片段',
timestamp BIGINT COMMENT '抓取时间戳',
content_hash VARCHAR(64) COMMENT '内容哈希(用于增量检测)',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
UNIQUE KEY uk_url (url(255)),
INDEX idx_url_hash (url_hash),
INDEX idx_timestamp (timestamp)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
COMMENT='URL摘要缓存表';
-- ============================================
-- 网站信息表
-- 存储每个域名的元信息
-- ============================================
CREATE TABLE IF NOT EXISTS site_info (
host VARCHAR(255) PRIMARY KEY COMMENT '域名/主机名',
visit_count INT UNSIGNED DEFAULT 0 COMMENT '访问次数',
last_visit_time BIGINT COMMENT '最后访问时间戳',
fingerprint TEXT COMMENT '网站指纹',
success_rate FLOAT COMMENT '成功率',
html_structure TEXT COMMENT 'HTML结构特征',
ips JSON COMMENT 'IP地址列表',
quality FLOAT COMMENT '质量评分',
https_available TINYINT COMMENT '是否支持HTTPS1=是,0=否)',
keywords JSON COMMENT '高频关键词列表',
out_links JSON COMMENT '出站链接列表',
languages JSON COMMENT '语种分布',
redirects JSON COMMENT '重定向链',
server_types JSON COMMENT 'Server类型',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
COMMENT='网站元信息表';
-- ============================================
-- 刷盘记录表
-- 用于断点续传,记录刷盘进度
-- ============================================
CREATE TABLE IF NOT EXISTS flush_marker (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
flush_type VARCHAR(50) NOT NULL COMMENT '刷盘类型:index, gate, site',
last_key VARCHAR(255) COMMENT '最后处理的key',
last_cursor BIGINT DEFAULT 0 COMMENT 'Redis SCAN游标',
processed_count INT UNSIGNED DEFAULT 0 COMMENT '本批次处理数量',
flush_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '刷盘时间',
INDEX idx_type (flush_type),
INDEX idx_time (flush_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
COMMENT='刷盘进度记录表';
-- ============================================
-- 优先爬取URL表
-- 存储需要优先爬取的URL
-- ============================================
CREATE TABLE IF NOT EXISTS priority_urls (
url VARCHAR(2048) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (url(255)),
INDEX idx_created (created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
COMMENT='优先爬取URL表';
+416
View File
@@ -0,0 +1,416 @@
// Package mysql 提供 MySQL 数据库连接和管理功能。
// 支持 Unix Socket 和 TCP 两种连接方式,自动初始化数据表和恢复数据。
package mysql
import (
"context"
"database/sql"
"fmt"
"log"
"os"
"path/filepath"
"strings"
"time"
goredis "github.com/redis/go-redis/v9"
_ "github.com/go-sql-driver/mysql"
"sese-engine/config"
)
// DB 是 MySQL 数据库连接池
var DB *sql.DB
// Open 初始化 MySQL 连接
// 根据配置自动选择 Unix Socket 或 TCP 连接
func Open() error {
dsn := config.MySQLDSN()
db, err := sql.Open("mysql", dsn)
if err != nil {
return fmt.Errorf("mysql.Open: %w", err)
}
// 配置连接池
db.SetConnMaxLifetime(time.Duration(config.MySQLConnMaxLifetime()) * time.Second)
db.SetMaxIdleConns(config.MySQLMaxIdleConns())
db.SetMaxOpenConns(config.MySQLMaxOpenConns())
// 验证连接
if err := db.Ping(); err != nil {
return fmt.Errorf("mysql.Ping: %w", err)
}
DB = db
log.Printf("[mysql] connected via %s", formatDSN(dsn))
// 自动初始化数据表
if err := initSchema(); err != nil {
return fmt.Errorf("mysql init schema: %w", err)
}
return nil
}
// initSchema 自动执行 init_db.sql 初始化数据表
func initSchema() error {
// 查找 init_db.sql 文件
execPath, err := os.Executable()
if err != nil {
execPath = os.Args[0]
}
sqlFile := filepath.Join(filepath.Dir(execPath), "mysql", "init_db.sql")
if _, err := os.Stat(sqlFile); os.IsNotExist(err) {
// 尝试从当前工作目录查找
cwd, _ := os.Getwd()
sqlFile = filepath.Join(cwd, "mysql", "init_db.sql")
}
data, err := os.ReadFile(sqlFile)
if err != nil {
return fmt.Errorf("read init_db.sql: %w", err)
}
// 获取配置的数据库名
dbName := config.Global.MySQL.Database
if dbName == "" {
dbName = "sese_engine"
}
log.Printf("[mysql] init schema: database=%s", dbName)
// 先切换到目标数据库
if _, err := DB.Exec("USE " + dbName); err != nil {
return fmt.Errorf("mysql USE database: %w", err)
}
// 分割 SQL 语句(按分号分割)
statements := splitStatements(string(data))
log.Printf("[mysql] found %d SQL statements to execute", len(statements))
execed := 0
for i, stmt := range statements {
trimmed := strings.TrimSpace(stmt)
// 跳过空行和注释
if trimmed == "" || strings.HasPrefix(trimmed, "--") || strings.HasPrefix(trimmed, "/*") {
log.Printf("[mysql] [%d/%d] SKIP (empty/comment): %s", i+1, len(statements), truncate(trimmed, 60))
continue
}
if _, err := DB.Exec(trimmed); err != nil {
log.Printf("[mysql] [%d/%d] FAILED: %v\n SQL: %s", i+1, len(statements), err, truncate(trimmed, 200))
continue
}
execed++
log.Printf("[mysql] [%d/%d] OK: %s", i+1, len(statements), truncate(trimmed, 60))
}
log.Printf("[mysql] init schema done, executed=%d statements", execed)
return nil
}
// splitStatements 按分号分割 SQL 语句(处理多行 CREATE TABLE
func splitStatements(sql string) []string {
var statements []string
var buf strings.Builder
inComment := false
for _, line := range strings.Split(sql, "\n") {
trimmed := strings.TrimSpace(line)
// 单行注释
if strings.HasPrefix(trimmed, "--") || strings.HasPrefix(trimmed, "//") {
continue
}
// 多行注释开始/结束
if strings.Contains(trimmed, "/*") {
inComment = true
}
if inComment {
if strings.Contains(trimmed, "*/") {
inComment = false
}
continue
}
// 空行跳过
if trimmed == "" {
continue
}
buf.WriteString(line)
buf.WriteString("\n")
// 检查是否以分号结尾
trimmed = strings.TrimSpace(buf.String())
if strings.HasSuffix(trimmed, ";") {
statements = append(statements, trimmed)
buf.Reset()
}
}
// 处理最后一条(可能没有分号)
if buf.Len() > 0 {
trimmed := strings.TrimSpace(buf.String())
if trimmed != "" {
statements = append(statements, trimmed)
}
}
return statements
}
// truncate 截断字符串
func truncate(s string, maxLen int) string {
if len(s) <= maxLen {
return s
}
return s[:maxLen] + "..."
}
// Close 关闭 MySQL 连接
func Close() error {
if DB != nil {
return DB.Close()
}
return nil
}
// Ping 检查 MySQL 连接是否正常
func Ping() error {
if DB == nil {
return fmt.Errorf("mysql not initialized")
}
return DB.Ping()
}
// formatDSN 格式化 DSN 用于日志(隐藏密码)
func formatDSN(dsn string) string {
// 简化日志输出
cfg := config.Global.MySQL
if cfg.UnixSocket != "" {
return fmt.Sprintf("unix_socket=%s database=%s", cfg.UnixSocket, cfg.Database)
}
return fmt.Sprintf("tcp=%s:%d database=%s", cfg.Host, cfg.Port, cfg.Database)
}
// RestoreFromMySQLToRedis 从 MySQL 恢复数据到 Redis
// 用于 Redis 数据丢失后重建索引
func RestoreFromMySQLToRedis(redisDB *goredis.Client) error {
if DB == nil {
return fmt.Errorf("mysql not initialized")
}
start := time.Now()
log.Printf("[mysql-restore] starting restoration from MySQL to Redis...")
ctx := context.Background()
// 1. 恢复 index_entries → Redis idx:* ZSet
if err := restoreIndexEntries(ctx, redisDB); err != nil {
return fmt.Errorf("restore index_entries: %w", err)
}
// 2. 恢复 url_snippets → Redis gate:* + url2hash:*
if err := restoreUrlSnippets(ctx, redisDB); err != nil {
return fmt.Errorf("restore url_snippets: %w", err)
}
// 3. 恢复 site_info → Redis site:*
if err := restoreSiteInfo(ctx, redisDB); err != nil {
return fmt.Errorf("restore site_info: %w", err)
}
// 4. 恢复 priority_urls → Redis priority:*
if err := restorePriorityURLs(ctx, redisDB); err != nil {
return fmt.Errorf("restore priority_urls: %w", err)
}
log.Printf("[mysql-restore] restoration completed in %v", time.Since(start))
return nil
}
// restoreIndexEntries 恢复倒排索引
func restoreIndexEntries(ctx context.Context, redisDB *goredis.Client) error {
rows, err := DB.Query("SELECT keyword, url, weight FROM index_entries")
if err != nil {
// 表不存在时跳过
log.Printf("[mysql-restore][index] skip: %v", err)
return nil
}
defer rows.Close()
// 按 keyword 分组
type indexRow struct {
URL string
Weight float32
}
keywordMap := make(map[string][]indexRow)
count := 0
for rows.Next() {
var keyword, url string
var weight float32
if err := rows.Scan(&keyword, &url, &weight); err != nil {
continue
}
keywordMap[keyword] = append(keywordMap[keyword], indexRow{URL: url, Weight: weight})
count++
}
// 批量写入 Redis
for keyword, entries := range keywordMap {
if len(entries) == 0 {
continue
}
zSlice := make([]goredis.Z, len(entries))
for i, e := range entries {
zSlice[i] = goredis.Z{Score: float64(e.Weight), Member: e.URL}
}
if err := redisDB.ZAdd(ctx, "idx:"+keyword, zSlice...).Err(); err != nil {
log.Printf("[mysql-restore][index] failed to restore %s: %v", keyword, err)
}
}
log.Printf("[mysql-restore][index] restored %d entries (%d keywords)", count, len(keywordMap))
return nil
}
// restoreUrlSnippets 恢复 URL 摘要
func restoreUrlSnippets(ctx context.Context, redisDB *goredis.Client) error {
rows, err := DB.Query("SELECT url, url_hash, title, description, text, timestamp, content_hash FROM url_snippets")
if err != nil {
// 表不存在时跳过
log.Printf("[mysql-restore][snippets] skip: %v", err)
return nil
}
defer rows.Close()
count := 0
for rows.Next() {
var url, urlHash, title, description, text, contentHash sql.NullString
var timestamp sql.NullInt64
if err := rows.Scan(&url, &urlHash, &title, &description, &text, &timestamp, &contentHash); err != nil {
continue
}
if !url.Valid || urlHash.Valid == false {
continue
}
fields := map[string]interface{}{
"url": url.String,
"title": nullString(title),
"desc": nullString(description),
"text": nullString(text),
"ts": nullInt64(timestamp),
"hash": nullString(contentHash),
}
if err := redisDB.HMSet(ctx, "gate:"+urlHash.String, fields).Err(); err != nil {
continue
}
// 同时写入 URL→hash 映射
redisDB.Set(ctx, "url2hash:"+url.String, urlHash.String, 0)
count++
}
log.Printf("[mysql-restore][snippets] restored %d entries", count)
return nil
}
// restoreSiteInfo 恢复网站信息
func restoreSiteInfo(ctx context.Context, redisDB *goredis.Client) error {
rows, err := DB.Query("SELECT host, visit_count, last_visit_time, success_rate, https_available FROM site_info")
if err != nil {
// 表不存在时跳过
log.Printf("[mysql-restore][site] skip: %v", err)
return nil
}
defer rows.Close()
count := 0
for rows.Next() {
var host string
var visitCount sql.NullInt64
var lastVisitTime sql.NullInt64
var successRate sql.NullFloat64
var httpsAvailable sql.NullInt64
if err := rows.Scan(&host, &visitCount, &lastVisitTime, &successRate, &httpsAvailable); err != nil {
continue
}
if host == "" {
continue
}
fields := map[string]interface{}{
"visit_count": nullInt64(visitCount),
"last_visit_time": nullInt64(lastVisitTime),
}
if successRate.Valid {
fields["success_rate"] = successRate.Float64
}
if httpsAvailable.Valid {
fields["https_available"] = httpsAvailable.Int64
}
if err := redisDB.HMSet(ctx, "site:"+host, fields).Err(); err != nil {
continue
}
count++
}
log.Printf("[mysql-restore][site] restored %d entries", count)
return nil
}
// restorePriorityURLs 恢复优先 URL
func restorePriorityURLs(ctx context.Context, redisDB *goredis.Client) error {
rows, err := DB.Query("SELECT url FROM priority_urls")
if err != nil {
// 表不存在时跳过
log.Printf("[mysql-restore][priority] skip: %v", err)
return nil
}
defer rows.Close()
count := 0
for rows.Next() {
var url string
if err := rows.Scan(&url); err != nil {
continue
}
if url == "" {
continue
}
fields := map[string]interface{}{
"url": url,
"is_domain": "0",
"added_at": time.Now().Unix(),
"visited": "0",
}
if err := redisDB.HMSet(ctx, "priority:"+url, fields).Err(); err != nil {
continue
}
count++
}
log.Printf("[mysql-restore][priority] restored %d entries", count)
return nil
}
// ---- 辅助函数 ----
func nullString(v sql.NullString) string {
if v.Valid {
return v.String
}
return ""
}
func nullInt64(v sql.NullInt64) int64 {
if v.Valid {
return v.Int64
}
return 0
}
+82
View File
@@ -0,0 +1,82 @@
// Package redis 提供 Redis 数据库连接和管理功能。
// 支持 Unix Socket 和 TCP 两种连接方式。
// 注意:此包作为纯内存缓存使用,不开启持久化。
package redis
import (
"context"
"fmt"
"log"
"time"
"github.com/redis/go-redis/v9"
"sese-engine/config"
)
// Client 是 Redis 客户端连接
var Client *redis.Client
// Open 初始化 Redis 连接
// 根据配置自动选择 Unix Socket 或 TCP 连接
// 注意:不开持久化,作为纯内存缓存使用
func Open() error {
cfg := config.Global.Redis
opt := &redis.Options{
Addr: config.RedisAddr(),
Password: config.RedisPassword(),
DB: config.RedisDB(),
PoolSize: config.RedisPoolSize(),
MinIdleConns: config.RedisMinIdleConns(),
ReadTimeout: time.Duration(config.RedisReadTimeout()) * time.Millisecond,
WriteTimeout: time.Duration(config.RedisWriteTimeout()) * time.Millisecond,
// 禁用持久化 - 作为纯内存缓存
// 不设置任何 save 策略即可禁用 RDB
}
client := redis.NewClient(opt)
Client = client
// 验证连接
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := client.Ping(ctx).Err(); err != nil {
return fmt.Errorf("redis.Ping: %w", err)
}
log.Printf("[redis] connected via %s (DB=%d)", formatAddr(opt), cfg.DB)
return nil
}
// Close 关闭 Redis 连接
func Close() error {
if Client != nil {
return Client.Close()
}
return nil
}
// Ping 检查 Redis 连接是否正常
func Ping() error {
if Client == nil {
return fmt.Errorf("redis not initialized")
}
return Client.Ping(context.Background()).Err()
}
// FlushDB 清空当前数据库(谨慎使用,仅用于测试或重置)
func FlushDB() error {
if Client == nil {
return fmt.Errorf("redis not initialized")
}
return Client.FlushDB(context.Background()).Err()
}
// formatAddr 格式化地址用于日志
func formatAddr(opt *redis.Options) string {
if opt.Addr != "" {
return opt.Addr
}
return fmt.Sprintf("%s:%d", opt.Addr, 6379)
}
+218 -60
View File
@@ -5,6 +5,7 @@ package search
import (
"container/heap" // 堆结构(域名交错排序)
"container/list" // 双向链表(LRU 缓存)
"context" // 超时控制(Redis 操作)
"encoding/json" // JSON 序列化(响应输出)
"fmt" // 错误格式化
"io" // 读取请求体
@@ -24,27 +25,34 @@ import (
"sync/atomic" // 原子操作(计数器)
"time" // 时间戳
goredis "github.com/redis/go-redis/v9" // Redis 客户端
"sese-engine/analyzer" // 分词和语言检测
"sese-engine/config" // 排序权重配置
"sese-engine/crawler" // 爬虫(读取活跃线程数)
"sese-engine/info" // info 服务
"sese-engine/parser" // HTML 解析(在线摘要)
sredis "sese-engine/redis" // 本地 Redis 连接(用于访问 sredis.Client
"sese-engine/storage" // 持久化存储
)
// urlKeywordsCache URL→关键词 的 LRU 缓存
// urlKeywordsCache URL→关键词 的 LRU 缓存(内存)+ Redis 镜像(TTL 2小时)
type urlKeywordsCache struct {
mu sync.RWMutex
items map[string]*list.Element // URL → list 节点
order *list.List // 按访问时间排序的双向链表
maxSize int
// Redis 双写
rdb *goredis.Client // Redis 客户端(懒初始化)
redisKey string // Redis Hash key
ttl time.Duration // TTL,默认 2 小时
}
// urlKeywordsEntry LRU 缓存条目
type urlKeywordsEntry struct {
URL string // URL(用于删除时从 map 中移除)
Title string // 页面标题(从 bbolt Snippet 缓存获取)
Snippet string // 摘要(从 bbolt Snippet 缓存获取)
Title string // 页面标题(从 Redis Snippet 缓存获取)
Snippet string // 摘要(从 Redis Snippet 缓存获取)
Keywords []urlKeywordInfo // 关键词列表
}
@@ -54,32 +62,40 @@ type urlKeywordInfo struct {
Weight float32 `json:"weight"` // 权重
}
// newURLKeywordsCache 创建一个新的 LRU 缓存
// newURLKeywordsCache 创建一个新的 LRU 缓存Redis 镜像通过 AttachRedis 注入)
func newURLKeywordsCache(maxSize int) *urlKeywordsCache {
return &urlKeywordsCache{
items: make(map[string]*list.Element),
order: list.New(),
maxSize: maxSize,
redisKey: "url_keywords:cache",
ttl: 2 * time.Hour,
}
}
// Put 写入或更新缓存
// AttachRedis 将 Redis 客户端注入缓存(用于双写)
func (c *urlKeywordsCache) AttachRedis(rdb *goredis.Client) {
c.rdb = rdb
}
// Put 写入或更新缓存(内存 LRU + Redis 双写,2小时 TTL
func (c *urlKeywordsCache) Put(url string, title, snippet string, keywords []urlKeywordInfo) {
c.mu.Lock()
defer c.mu.Unlock()
var entry *urlKeywordsEntry
// 已存在,移到队尾(更新新鲜度)
if elem, ok := c.items[url]; ok {
c.order.MoveToBack(elem)
entry := elem.Value.(*urlKeywordsEntry)
entry.Keywords = keywords
entry.Title = title
entry.Snippet = snippet
return
}
e := elem.Value.(*urlKeywordsEntry)
e.Keywords = keywords
e.Title = title
e.Snippet = snippet
entry = e
} else {
// 新增到队尾
entry := &urlKeywordsEntry{URL: url, Title: title, Snippet: snippet, Keywords: keywords}
entry = &urlKeywordsEntry{URL: url, Title: title, Snippet: snippet, Keywords: keywords}
elem := c.order.PushBack(entry)
c.items[url] = elem
@@ -91,65 +107,205 @@ func (c *urlKeywordsCache) Put(url string, title, snippet string, keywords []url
delete(c.items, oldest.Value.(*urlKeywordsEntry).URL)
}
}
}
// Redis 双写(异步,不阻塞主流程)
go c.redisPut(url, entry)
}
// Get 读取缓存,同时更新新鲜度
func (c *urlKeywordsCache) Get(url string) (*urlKeywordsEntry, bool) {
c.mu.Lock()
defer c.mu.Unlock()
// redisPut 将条目写入 Redis HashTTL 2小时)
func (c *urlKeywordsCache) redisPut(url string, entry *urlKeywordsEntry) {
if c.rdb == nil {
return
}
data, err := json.Marshal(entry)
if err != nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
pipe := c.rdb.Pipeline()
pipe.HSet(ctx, c.redisKey, url, string(data))
pipe.Expire(ctx, c.redisKey, c.ttl)
pipe.Exec(ctx)
}
if elem, ok := c.items[url]; ok {
c.order.MoveToBack(elem) // 访问过,移到队尾
// Get 读取缓存(内存优先,Redis 回填)
func (c *urlKeywordsCache) Get(url string) (*urlKeywordsEntry, bool) {
// 先查内存
c.mu.RLock()
elem, ok := c.items[url]
if ok {
entry := elem.Value.(*urlKeywordsEntry)
c.mu.RUnlock()
// 异步更新 Redis TTL
go c.touchRedisTTL(url)
return entry, true
}
c.mu.RUnlock()
// 内存 miss,查 Redis
entry, ok := c.redisGet(url)
if !ok {
return nil, false
}
// 回填内存(写锁)
c.mu.Lock()
// 二次检查(可能已有其他协程写入)
if _, exists := c.items[url]; !exists {
// 超过上限时,先删最旧的再加入
if c.order.Len() >= c.maxSize {
oldest := c.order.Front()
if oldest != nil {
c.order.Remove(oldest)
delete(c.items, oldest.Value.(*urlKeywordsEntry).URL)
}
}
elem := c.order.PushBack(entry)
c.items[url] = elem
}
c.mu.Unlock()
return entry, true
}
// Stats 返回缓存统计信息
// redisGet 从 Redis Hash 获取单条缓存
func (c *urlKeywordsCache) redisGet(url string) (*urlKeywordsEntry, bool) {
if c.rdb == nil {
return nil, false
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
data, err := c.rdb.HGet(ctx, c.redisKey, url).Result()
if err != nil {
return nil, false
}
var entry urlKeywordsEntry
if err := json.Unmarshal([]byte(data), &entry); err != nil {
return nil, false
}
// 续命 Redis TTL
go c.touchRedisTTL(url)
return &entry, true
}
// touchRedisTTL 续命 Redis 条目 TTL
func (c *urlKeywordsCache) touchRedisTTL(url string) {
if c.rdb == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
c.rdb.Expire(ctx, c.redisKey, c.ttl)
}
// Stats 返回缓存统计信息(优先 Redis 总数,包含 TTL 内的冷数据)
func (c *urlKeywordsCache) Stats() (size int, maxSize int) {
c.mu.RLock()
defer c.mu.RUnlock()
return len(c.items), c.maxSize
}
memSize := len(c.items)
maxSize = c.maxSize
c.mu.RUnlock()
// ListAll 返回所有缓存条目列表(按访问时间从旧到新
func (c *urlKeywordsCache) ListAll() []*urlKeywordsEntry {
c.mu.RLock()
defer c.mu.RUnlock()
entries := make([]*urlKeywordsEntry, 0, len(c.items))
for elem := c.order.Front(); elem != nil; elem = elem.Next() {
entries = append(entries, elem.Value.(*urlKeywordsEntry))
// 尝试获取 Redis 总数(更准确,包含未淘汰的冷数据
if c.rdb != nil {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
n, err := c.rdb.HLen(ctx, c.redisKey).Result()
if err == nil && n > 0 {
return int(n), maxSize
}
return entries
}
return memSize, maxSize
}
// ListPage 返回分页缓存条目(按访问时间从新到旧,最新访问的在前
func (c *urlKeywordsCache) ListPage(page, pageSize int) []*urlKeywordsEntry {
// ListAll 返回所有缓存条目列表(内存 + Redis 冷数据合并
func (c *urlKeywordsCache) ListAll() []*urlKeywordsEntry {
// 收集内存数据
c.mu.RLock()
defer c.mu.RUnlock()
total := len(c.items)
memEntries := make([]*urlKeywordsEntry, 0, len(c.items))
for elem := c.order.Front(); elem != nil; elem = elem.Next() {
memEntries = append(memEntries, elem.Value.(*urlKeywordsEntry))
}
c.mu.RUnlock()
// 尝试从 Redis 获取补充数据
if c.rdb == nil {
return memEntries
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
all, err := c.rdb.HGetAll(ctx, c.redisKey).Result()
if err != nil || len(all) == 0 {
return memEntries
}
// 去重合并(内存优先)
seen := make(map[string]bool, len(memEntries))
for _, e := range memEntries {
seen[e.URL] = true
}
for _, v := range all {
var entry urlKeywordsEntry
if err := json.Unmarshal([]byte(v), &entry); err != nil {
continue
}
if !seen[entry.URL] {
memEntries = append(memEntries, &entry)
seen[entry.URL] = true
}
}
return memEntries
}
// ListPage 返回分页缓存条目(内存 + Redis 冷数据合并,按 URL 字典序排列)
func (c *urlKeywordsCache) ListPage(page, pageSize int) []*urlKeywordsEntry {
// 收集内存数据
c.mu.RLock()
memEntries := make([]*urlKeywordsEntry, 0, len(c.items))
for elem := c.order.Front(); elem != nil; elem = elem.Next() {
memEntries = append(memEntries, elem.Value.(*urlKeywordsEntry))
}
c.mu.RUnlock()
// 从 Redis 补充冷数据
if c.rdb != nil {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
all, err := c.rdb.HGetAll(ctx, c.redisKey).Result()
if err == nil && len(all) > 0 {
seen := make(map[string]bool, len(memEntries))
for _, e := range memEntries {
seen[e.URL] = true
}
for _, v := range all {
var entry urlKeywordsEntry
if err := json.Unmarshal([]byte(v), &entry); err != nil {
continue
}
if !seen[entry.URL] {
memEntries = append(memEntries, &entry)
seen[entry.URL] = true
}
}
}
}
// 分页(从合并数据中按字典序取 page 页)
total := len(memEntries)
offset := (page - 1) * pageSize
if offset >= total {
return []*urlKeywordsEntry{}
}
// 从 Back(最新)向前遍历,跳过 offset 个
elem := c.order.Back()
for i := 0; i < offset && elem != nil; i++ {
elem = elem.Prev()
end := offset + pageSize
if end > total {
end = total
}
// 收集 pageSize 条
entries := make([]*urlKeywordsEntry, 0, pageSize)
for i := 0; i < pageSize && elem != nil; i++ {
entries = append(entries, elem.Value.(*urlKeywordsEntry))
elem = elem.Prev()
}
return entries
return memEntries[offset:end]
}
// Server 是搜索 HTTP 服务器,同时内嵌收获服务(统一在同一端口)。
type Server struct {
db *storage.DB
db *storage.RedisStoreV2
infoSvc *info.Service
analyzer *analyzer.Analyzer
httpCli *http.Client // 在线摘要抓取(无 robots.txt 检查)
@@ -166,11 +322,11 @@ type Server struct {
indexCacheMu sync.RWMutex
indexCacheHits int64 // 缓存命中计数(原子)
// stats 快照缓存:后台定时刷新,避免每次请求全量遍历 bbolt
// stats 快照缓存:后台定时刷新,避免每次请求全量遍历 Redis
statsCache map[string]any
statsCacheMu sync.RWMutex
// recent 快照缓存:后台定时刷新,避免每次请求全量遍历 bbolt
// recent 快照缓存:后台定时刷新,避免每次请求全量遍历 Redis
recentCache map[int][]recentItem // limit → 预截取的结果列表
recentCacheMu sync.RWMutex
recentTotal int // 总条目数(不截取)
@@ -192,7 +348,7 @@ type Server struct {
}
// New 创建一个 search Server(内嵌收获服务,统一在同一端口)。
func New(db *storage.DB, infoSvc *info.Service, a *analyzer.Analyzer) *Server {
func New(db *storage.RedisStoreV2, infoSvc *info.Service, a *analyzer.Analyzer) *Server {
s := &Server{
db: db,
infoSvc: infoSvc,
@@ -203,6 +359,8 @@ func New(db *storage.DB, infoSvc *info.Service, a *analyzer.Analyzer) *Server {
},
urlKeywords: newURLKeywordsCache(10000), // 缓存 1 万个 URL 的关键词
}
// 注入 Redis 客户端到 urlKeywords 缓存(用于双写,2小时 TTL)
s.urlKeywords.AttachRedis(sredis.Client)
// 启动定期刷盘 goroutine
go s.runPeriodicFlush()
// 启动 stats + recent 缓存定期刷新 goroutine
@@ -383,7 +541,7 @@ type recentItem struct {
}
// handleAdminRecent 返回最近爬取的条目列表,按爬取时间倒序。
// 直接返回缓存快照,不阻塞 bbolt,响应时间 <1ms。
// 直接返回缓存快照,不阻塞 Redis,响应时间 <1ms。
// 参数:limit(默认50,最大200)。
func (s *Server) handleAdminRecent(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Access-Control-Allow-Origin", "*")
@@ -419,7 +577,7 @@ func (s *Server) handleAdminRecent(w http.ResponseWriter, r *http.Request) {
})
}
// refreshRecentCache 全量遍历 bbolt 计算 recent 快照,预截取常用 limit,存入 recentCache。
// refreshRecentCache 全量遍历 Redis 计算 recent 快照,预截取常用 limit,存入 recentCache。
func (s *Server) refreshRecentCache() {
type entry struct {
url string
@@ -495,7 +653,7 @@ func (s *Server) refreshRecentCache() {
}
// handleAdminStats 返回全局统计:域名分布、语种分布、总 URL 数、总词数。
// 直接返回缓存快照,不阻塞 bbolt,响应时间 <1ms。
// 直接返回缓存快照,不阻塞 Redis,响应时间 <1ms。
func (s *Server) handleAdminStats(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Content-Type", "application/json; charset=utf-8")
@@ -523,7 +681,7 @@ func (s *Server) handleAdminStats(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(cached)
}
// refreshStatsCache 全量遍历 bbolt 计算统计快照,存入 statsCache。
// refreshStatsCache 全量遍历 Redis 计算统计快照,存入 statsCache。
func (s *Server) refreshStatsCache() {
domainCount := make(map[string]int)
langCount := make(map[string]int)
@@ -556,7 +714,7 @@ func (s *Server) refreshStatsCache() {
return nil
})
// 遍历结束后批量查 SiteInfo(避免 ForEachSnippet 回调中嵌套 bbolt 事务)
// 遍历结束后批量查 SiteInfo(避免 ForEachSnippet 回调中嵌套 Redis 事务)
for _, domain := range snippetDomains {
siteInfo, _ := s.db.GetSiteInfo(domain)
if siteInfo != nil {
@@ -616,7 +774,7 @@ func (s *Server) refreshStatsCache() {
}
// runCacheRefresher 后台定时刷新 stats 和 recent 缓存。
// 统一由一个 goroutine 交替刷新,避免同时全量遍历 bbolt 造成压力。
// 统一由一个 goroutine 交替刷新,避免同时全量遍历 Redis 造成压力。
func (s *Server) runCacheRefresher() {
interval := time.Duration(config.StatsRefreshInterval()) * time.Second
ticker := time.NewTicker(interval)
@@ -1487,8 +1645,8 @@ func timeMul(si *storage.SiteInfo, now int64) float64 {
}
// urlTimeMul 根据该 URL 的摘要抓取时间计算时间衰减倍数(30 天内不衰减)。
func urlTimeMul(db *storage.DB, rawURL string, now int64) float64 {
entry, err := db.GetSnippet(rawURL)
func urlTimeMul(store *storage.RedisStoreV2, rawURL string, now int64) float64 {
entry, err := store.GetSnippet(rawURL)
if err != nil || entry == nil {
return 1.0
}
@@ -1941,7 +2099,7 @@ func (s *Server) handleIngest(w http.ResponseWriter, r *http.Request) {
go s.Flush()
}
// 更新 URL→关键词 LRU 缓存(从 bbolt 获取标题和摘要)
// 更新 URL→关键词 LRU 缓存(从 Redis 获取标题和摘要)
keywords := make([]urlKeywordInfo, len(payload.Keywords))
for i, kw := range payload.Keywords {
keywords[i] = urlKeywordInfo{
@@ -1950,7 +2108,7 @@ func (s *Server) handleIngest(w http.ResponseWriter, r *http.Request) {
}
}
title, snippet := "", ""
if snippetEntry, err := s.db.GetSnippet(payload.URL); err == nil {
if snippetEntry, err := s.db.GetSnippet(payload.URL); err == nil && snippetEntry != nil {
title = snippetEntry.Title
snippet = snippetEntry.Description
if snippet == "" {
-750
View File
@@ -1,750 +0,0 @@
// Package storage provides the persistent index and site-info storage backed by bbolt.
// storage 包提供基于 bbolt 的持久化存储,负责保存倒排索引、URL摘要缓存和网站元信息。
//
// 索引空间(index bucket):key = 关键词(string),value = brotli 压缩的 JSON 数组,每项为 [权重, URL] 对。
// 融合之门(gate bucket):key = URLstring),value = brotli 压缩的 JSON 数组 [标题, 描述, 正文, 时间戳]。
// 网站之门(site_gate bucket):key = 主机名(string),value = brotli 压缩的 JSON SiteInfo 结构体。
//
// Python 版使用自定义哈希桶结构;Go 版直接交由 bbolt 原生处理。
package storage
import (
"encoding/json" // JSON 序列化/反序列化
"fmt" // 格式化错误信息
"io" // io.EOF 常量
"log" // 日志输出
"os" // 操作系统功能(创建目录等)
"path/filepath" // 路径拼接
"sync" // 互斥锁(保护写缓冲)
"time" // bbolt 超时配置和写缓冲定时器
"github.com/andybalholm/brotli" // Brotli 无损压缩库(用于压缩存储数据)
bolt "go.etcd.io/bbolt" // BoltDB,纯 Go 嵌入式 KV 数据库
)
// IndexEntry 是倒排索引中的单个条目。
// 一条索引记录表示"某个 URL 与某个关键词的相关性权重"。
type IndexEntry struct {
Weight float32 `json:"w"` // 该 URL 在该关键词下的得分/权重
URL string `json:"u"` // 网页 URL
}
// SnippetEntry 是 URL 对应的摘要信息缓存。
// 包含页面标题、描述、正文片段、抓取时间戳和内容哈希(用于增量重爬检测)。
type SnippetEntry struct {
Title string `json:"title"` // 网页标题
Description string `json:"desc"` // meta description 或自动生成的描述
Text string `json:"text"` // 正文前 N 字符的文本片段
Timestamp int64 `json:"ts"` // 抓取该页面时的 Unix 时间戳
ContentHash string `json:"hash"` // 正文内容的 FNV-1a 哈希(用于增量重爬判断内容是否变化)
}
// 四个 bbolt bucket 的名称(以字节数组存储,bbolt 要求 key/value 均为字节)
var (
bucketIndex = []byte("index") // 倒排索引 bucket
bucketGate = []byte("gate") // URL 摘要缓存 bucket
bucketSiteGate = []byte("site_gate") // 网站元信息 bucket
bucketPriority = []byte("priority") // 优先爬取 URL bucket
)
// writeOp 表示一个待写入的 snippet 操作。
type writeOp struct {
data []byte // marshalCompress 后的数据
}
// DB 封装一个 bbolt 数据库,提供类型化的存取接口。
// bbolt 内部已实现并发安全,无需额外加锁。
type DB struct {
db *bolt.DB // 底层 bbolt 数据库句柄
// 异步写缓冲:SetSnippet 先写到内存,定期批量刷入 bbolt。
writeBuf map[string]*writeOp // key → 待写入的操作
writeBufMu sync.Mutex
writeTicker *time.Ticker
writeDone chan struct{}
// SiteInfo 独立内存缓存:读全走缓存(零 bbolt 读事务),定期批量刷入 bbolt。
// 使用 RWMutex 允许多个读并发,写(UpdateSiteInfo)独占。
siteCache map[string]*SiteInfo // host → 最新 SiteInfo
siteCacheMu sync.RWMutex
siteDirty map[string]struct{} // 需要刷盘的 host 集合
siteDirtyMu sync.Mutex
siteTicker *time.Ticker
siteDone chan struct{}
}
// Open 在指定目录路径下创建或打开 bbolt 数据库文件。
// 如果目录不存在会自动创建。数据库文件名为 sese.db。
func Open(dir string) (*DB, error) {
// 确保存储目录存在(0775 权限:所有者读写执行,组用户读执行,其他读执行)
if err := os.MkdirAll(dir, 0o755); err != nil {
return nil, fmt.Errorf("storage.Open mkdir: %w", err)
}
// 拼接数据库文件路径:dir/sese.db
path := filepath.Join(dir, "sese.db")
// 打开/创建数据库文件,文件权限 0600(仅所有者可读写)
// NoSync: true — 不在每次写事务后 fsync,交由 OS 决定刷盘时机。
// 在高并发写入场景下大幅减少磁盘 I/O 阻塞,代价是极端断电可能丢失最近几秒数据(可接受)。
db, err := bolt.Open(path, 0o600, &bolt.Options{
NoSync: true,
Timeout: 5 * time.Second,
PageSize: 4096,
})
if err != nil {
return nil, fmt.Errorf("storage.Open bolt: %w", err)
}
// 启动时确保四个 bucket 都存在(不存在则创建)
err = db.Update(func(tx *bolt.Tx) error {
for _, b := range [][]byte{bucketIndex, bucketGate, bucketSiteGate, bucketPriority} {
if _, err := tx.CreateBucketIfNotExists(b); err != nil {
return err
}
}
return nil
})
if err != nil {
return nil, fmt.Errorf("storage.Open create buckets: %w", err)
}
d := &DB{db: db}
// 启动时预热 SiteInfo 缓存:一次性从 bbolt 加载所有 SiteInfo 到内存
d.siteCache = make(map[string]*SiteInfo)
_ = db.View(func(tx *bolt.Tx) error {
return tx.Bucket(bucketSiteGate).ForEach(func(k, v []byte) error {
var info SiteInfo
if err := decompressUnmarshal(v, &info); err != nil {
return nil // 跳过损坏条目
}
if info.Languages == nil {
info.Languages = make(map[string]float64)
}
if info.Redirects == nil {
info.Redirects = make(map[string]string)
}
d.siteCache[string(k)] = &info
return nil
})
})
log.Printf("[storage] siteCache warmed: %d hosts loaded", len(d.siteCache))
return d, nil
}
// StartWriteFlusher 启动后台写缓冲定时刷盘 goroutinesnippet 和 SiteInfo 各自独立)。
func (d *DB) StartWriteFlusher() {
// Snippet 写缓冲
d.writeBuf = make(map[string]*writeOp)
d.writeTicker = time.NewTicker(2 * time.Second)
d.writeDone = make(chan struct{})
go func() {
for {
select {
case <-d.writeTicker.C:
d.flushWriteBuf()
case <-d.writeDone:
return
}
}
}()
// SiteInfo 独立缓存
d.siteCache = make(map[string]*SiteInfo)
d.siteDirty = make(map[string]struct{})
d.siteTicker = time.NewTicker(5 * time.Second)
d.siteDone = make(chan struct{})
go func() {
for {
select {
case <-d.siteTicker.C:
d.flushSiteCache()
case <-d.siteDone:
return
}
}
}()
}
// flushWriteBuf 将写缓冲中的 snippet 操作批量刷入 bbolt。
func (d *DB) flushWriteBuf() {
d.writeBufMu.Lock()
if len(d.writeBuf) == 0 {
d.writeBufMu.Unlock()
return
}
// 快照并清空缓冲
snapshot := d.writeBuf
d.writeBuf = make(map[string]*writeOp)
d.writeBufMu.Unlock()
// 单个事务批量写入
if err := d.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucketGate)
for k, op := range snapshot {
if err := b.Put([]byte(k), op.data); err != nil {
return err
}
}
return nil
}); err != nil {
log.Printf("[storage] flushWriteBuf error: %v", err)
}
}
// flushSiteCache 将脏的 SiteInfo 批量刷入 bbolt。
func (d *DB) flushSiteCache() {
d.siteDirtyMu.Lock()
if len(d.siteDirty) == 0 {
d.siteDirtyMu.Unlock()
return
}
dirty := d.siteDirty
d.siteDirty = make(map[string]struct{})
d.siteDirtyMu.Unlock()
// 在读锁下快照所有脏数据并预压缩
d.siteCacheMu.RLock()
type kv struct {
host string
data []byte
}
items := make([]kv, 0, len(dirty))
for host := range dirty {
if info, ok := d.siteCache[host]; ok {
data, err := marshalCompress(info)
if err != nil {
log.Printf("[storage] flushSiteCache marshal error for %s: %v", host, err)
continue
}
items = append(items, kv{host, data})
}
}
d.siteCacheMu.RUnlock()
if len(items) == 0 {
return
}
// 单个事务批量写入 bbolt
if err := d.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucketSiteGate)
for _, item := range items {
if err := b.Put([]byte(item.host), item.data); err != nil {
return err
}
}
return nil
}); err != nil {
log.Printf("[storage] flushSiteCache error: %v", err)
}
}
// Close 关闭底层 bbolt 数据库连接。
// 先刷完写缓冲和 SiteInfo 缓存,再关闭定时器,最后关闭数据库。
func (d *DB) Close() error {
if d.writeTicker != nil {
d.writeTicker.Stop()
}
if d.siteTicker != nil {
d.siteTicker.Stop()
}
d.flushWriteBuf()
d.flushSiteCache()
if d.writeDone != nil {
close(d.writeDone)
}
if d.siteDone != nil {
close(d.siteDone)
}
return d.db.Close()
}
// ---- 辅助函数:压缩与解压、map 拷贝 ----
// copyMapF64 深拷贝 map[string]float64。
func copyMapF64(m map[string]float64) map[string]float64 {
cp := make(map[string]float64, len(m))
for k, v := range m {
cp[k] = v
}
return cp
}
// copyMapStr 深拷贝 map[string]string。
func copyMapStr(m map[string]string) map[string]string {
cp := make(map[string]string, len(m))
for k, v := range m {
cp[k] = v
}
return cp
}
// compress 将字节数组用 brotli 压缩后返回。
// brotli 压缩比高于 gzip,适合大量文本的存储空间优化。
func compress(data []byte) ([]byte, error) {
buf := make([]byte, 0, len(data)) // 预分配,避免反复扩容
w := brotli.NewWriterLevel((*appendWriter)(&buf), 3) // 压缩级别 3(优先速度,压缩比损失约 10-15%)
if _, err := w.Write(data); err != nil {
return nil, err
}
if err := w.Close(); err != nil {
return nil, err
}
return buf, nil
}
// decompress 将 brotli 压缩的字节数组解压还原。
func decompress(data []byte) ([]byte, error) {
// brotli.NewReader 从字节数组创建读取器(通过 byteReader 适配 io.Reader 接口)
r := brotli.NewReader(
(*byteReader)(&data),
)
out := make([]byte, 0, len(data)*3) // 预分配约 3 倍空间(解压后通常更大)
tmp := make([]byte, 4096) // 每次最多读 4KB
for {
n, err := r.Read(tmp)
out = append(out, tmp[:n]...) // 追加本次读取的字节
if err != nil {
if err == io.EOF {
break // 正常读完
}
return out, err // 其他错误(非 EOF)则返回
}
}
return out, nil
}
// appendWriter 将 *[]byte 适配为 io.Writer 接口(写入时直接 append)。
type appendWriter []byte
// Write 将数据 p 追加到 appendWriter 末尾,返回写入字节数。
func (a *appendWriter) Write(p []byte) (int, error) {
*a = append(*a, p...)
return len(p), nil
}
// byteReader 将 []byte 适配为 io.Reader 接口(顺序读取,支持读完后返回 EOF)。
type byteReader []byte
// Read 从字节数组读取最多 len(p) 字节到 p 中,返回实际读取字节数和可能的错误。
// 当字节数组全部读完后返回 io.EOF。
func (b *byteReader) Read(p []byte) (int, error) {
if len(*b) == 0 {
return 0, io.EOF // 已读完
}
n := copy(p, *b) // 复制最多 len(p) 字节
*b = (*b)[n:] // 前进指针
return n, nil
}
// marshalCompress 将任意可序列化对象先 JSON 编码,再 brotli 压缩,返回压缩后的字节。
func marshalCompress(v any) ([]byte, error) {
raw, err := json.Marshal(v) // 先序列化为 JSON
if err != nil {
return nil, err
}
return compress(raw) // 再压缩
}
// decompressUnmarshal 将压缩字节先解压,再 JSON 反序列化到目标对象 v。
func decompressUnmarshal(data []byte, v any) error {
raw, err := decompress(data) // 先解压
if err != nil {
return err
}
return json.Unmarshal(raw, v) // 再反序列化
}
// ---- 倒排索引(Index)相关方法 ----
// GetIndex 根据关键词查询倒排索引,返回该词关联的所有 [权重, URL] 条目列表。
func (d *DB) GetIndex(keyword string) ([]IndexEntry, error) {
var entries []IndexEntry
err := d.db.View(func(tx *bolt.Tx) error {
b := tx.Bucket(bucketIndex)
v := b.Get([]byte(keyword)) // 在 index bucket 中按关键词查询
if v == nil {
return nil // 不存在该词,返回空列表
}
return decompressUnmarshal(v, &entries)
})
return entries, err
}
// SetIndex 将某关键词的完整索引条目列表覆盖写入(替换旧数据)。
func (d *DB) SetIndex(keyword string, entries []IndexEntry) error {
data, err := marshalCompress(entries)
if err != nil {
return err
}
return d.db.Update(func(tx *bolt.Tx) error {
return tx.Bucket(bucketIndex).Put([]byte(keyword), data)
})
}
// BatchSetIndex 分批写入多个关键词→条目列表的映射。
// 大数据量时拆分为多个小事务,避免单个大事务导致的长时间阻塞。
func (d *DB) BatchSetIndex(batch map[string][]IndexEntry) error {
const batchSize = 1000
items := make([]struct {
keyword string
entries []IndexEntry
}, 0, len(batch))
for k, v := range batch {
items = append(items, struct {
keyword string
entries []IndexEntry
}{k, v})
}
totalBatches := (len(items) + batchSize - 1) / batchSize
for i := 0; i < len(items); i += batchSize {
end := i + batchSize
if end > len(items) {
end = len(items)
}
batchNum := i/batchSize + 1
// 事务外预先完成所有序列化和压缩,减少事务持锁时间
preItems := make([]struct {
keyword string
data []byte
}, 0, end-i)
for _, item := range items[i:end] {
data, err := marshalCompress(item.entries)
if err != nil {
return err
}
preItems = append(preItems, struct {
keyword string
data []byte
}{item.keyword, data})
}
// 事务内只做纯内存写入,持锁时间极短
if err := d.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(bucketIndex)
for _, p := range preItems {
if err := b.Put([]byte(p.keyword), p.data); err != nil {
return err
}
}
return nil
}); err != nil {
return err
}
if totalBatches > 1 {
log.Printf("[storage] BatchSetIndex progress: batch %d/%d (%d keys)", batchNum, totalBatches, end)
}
}
return nil
}
// ForEachIndex 遍历倒排索引中所有关键词及其关联条目,对每个条目调用 fn。
// 用于全量读取索引、做备份或重新计算等场景。
func (d *DB) ForEachIndex(fn func(keyword string, entries []IndexEntry) error) error {
return d.db.View(func(tx *bolt.Tx) error {
return tx.Bucket(bucketIndex).ForEach(func(k, v []byte) error {
var entries []IndexEntry
if err := decompressUnmarshal(v, &entries); err != nil {
return nil // 跳过损坏条目,不中断遍历
}
return fn(string(k), entries)
})
})
}
// ---- 融合之门(Gate):URL 摘要缓存相关方法 ----
// GetSnippet 根据 URL 查询缓存的摘要信息(标题/描述/正文片段)。
// 若未命中返回 error。
func (d *DB) GetSnippet(url string) (*SnippetEntry, error) {
var entry SnippetEntry
err := d.db.View(func(tx *bolt.Tx) error {
v := tx.Bucket(bucketGate).Get([]byte(url))
if v == nil {
return fmt.Errorf("not found")
}
return decompressUnmarshal(v, &entry)
})
if err != nil {
return nil, err
}
return &entry, nil
}
// SetSnippet 将某 URL 的摘要信息写入写缓冲(异步批量刷入磁盘)。
func (d *DB) SetSnippet(url string, entry *SnippetEntry) error {
data, err := marshalCompress(entry)
if err != nil {
return err
}
d.writeBufMu.Lock()
d.writeBuf[url] = &writeOp{data: data}
// 缓冲过大时同步刷一次,防止内存膨胀
if len(d.writeBuf) >= 5000 {
d.writeBufMu.Unlock()
d.flushWriteBuf()
return nil
}
d.writeBufMu.Unlock()
return nil
}
// ---- 网站之门(SiteGate):网站元信息相关方法 ----
// SiteInfo 存放每个域名/主机的元信息,与 Python 版网站.py 的 dataclass 对应。
type SiteInfo struct {
VisitCount int `json:"visit_count"` // 累计访问该网站的次数
LastVisitTime int64 `json:"last_visit_time"` // 上次访问该网站的时间戳
Fingerprint any `json:"fingerprint,omitempty"` // 网站指纹(用于识别重复站点)
SuccessRate *float64 `json:"success_rate,omitempty"` // 访问成功率(成功次数/总访问次数)
HTMLStructure string `json:"html_structure,omitempty"` // HTML 结构特征摘要
IPs []string `json:"ips,omitempty"` // 该域名解析出的 IP 列表
Quality *float64 `json:"quality,omitempty"` // 网站质量评分(0~1
HTTPSAvailable *bool `json:"https_available,omitempty"` // 是否支持 HTTPS
Keywords []string `json:"keywords,omitempty"` // 该网站的高频关键词列表
OutLinks []string `json:"out_links,omitempty"` // 从该网站页面提取的出站链接列表
Languages map[string]float64 `json:"languages,omitempty"` // 网站语种分布(语种代码 → 占比)
Redirects map[string]string `json:"redirects,omitempty"` // 重定向链(URL → 最终 URL)
ServerTypes []string `json:"server_types,omitempty"` // 网站使用的 HTTP Server 类型列表
}
// GetSiteInfo 根据主机名查询网站元信息。
// 全程只读内存缓存(零 bbolt 读事务),适合高并发调用。
// 若不存在则返回仅有默认空 map 的空 SiteInfo(不报错,方便调用方直接使用)。
func (d *DB) GetSiteInfo(host string) (*SiteInfo, error) {
d.siteCacheMu.RLock()
info, ok := d.siteCache[host]
d.siteCacheMu.RUnlock()
if ok {
// 返回深拷贝,防止调用方修改缓存
cp := *info
if cp.Languages == nil {
cp.Languages = make(map[string]float64)
} else {
cp.Languages = copyMapF64(cp.Languages)
}
if cp.Redirects == nil {
cp.Redirects = make(map[string]string)
} else {
cp.Redirects = copyMapStr(cp.Redirects)
}
return &cp, nil
}
// 缓存未命中,从 bbolt 加载
// 注意:不能在 bbolt 事务回调(如 ForEachSnippet 的 fn)内调用此路径(bbolt 不支持嵌套事务)。
// 正常情况下预热已加载全部数据,不会走到这里。
var si SiteInfo
var err error
func() {
defer func() {
if r := recover(); r != nil {
// bbolt 嵌套事务 panic,返回空 SiteInfo
si = SiteInfo{Languages: make(map[string]float64), Redirects: make(map[string]string)}
err = nil
}
}()
err = d.db.View(func(tx *bolt.Tx) error {
v := tx.Bucket(bucketSiteGate).Get([]byte(host))
if v == nil {
return fmt.Errorf("not found")
}
return decompressUnmarshal(v, &si)
})
}()
if err != nil {
return &SiteInfo{Languages: make(map[string]float64), Redirects: make(map[string]string)}, nil
}
if si.Languages == nil {
si.Languages = make(map[string]float64)
}
if si.Redirects == nil {
si.Redirects = make(map[string]string)
}
// 回填缓存
d.siteCacheMu.Lock()
if existing, exists := d.siteCache[host]; !exists {
d.siteCache[host] = &si
} else {
si = *existing // 用更新的版本
}
d.siteCacheMu.Unlock()
return &si, nil
}
// SetSiteInfo 将网站元信息直接写入内存缓存并标记为脏(异步刷入磁盘)。
func (d *DB) SetSiteInfo(host string, info *SiteInfo) error {
cp := *info // 浅拷贝,断开与调用方的引用
if cp.Languages != nil {
cp.Languages = copyMapF64(cp.Languages)
}
if cp.Redirects != nil {
cp.Redirects = copyMapStr(cp.Redirects)
}
d.siteCacheMu.Lock()
d.siteCache[host] = &cp
d.siteCacheMu.Unlock()
d.siteDirtyMu.Lock()
d.siteDirty[host] = struct{}{}
d.siteDirtyMu.Unlock()
return nil
}
// UpdateSiteInfo 原子地读取当前 SiteInfo 并应用修改函数 fn,然后写回缓存。
// 只用 siteCacheMu(不碰 bbolt),不会与 flush 产生跨锁依赖。
func (d *DB) UpdateSiteInfo(host string, fn func(*SiteInfo)) error {
d.siteCacheMu.Lock()
// 从缓存读取(缓存未命中会返回 nil,后续初始化为空)
info, ok := d.siteCache[host]
if !ok {
// 从 bbolt 加载(注意:这是唯一会做 bbolt 读的地方,但只在首次 miss 时触发)
var si SiteInfo
err := d.db.View(func(tx *bolt.Tx) error {
v := tx.Bucket(bucketSiteGate).Get([]byte(host))
if v == nil {
return fmt.Errorf("not found")
}
return decompressUnmarshal(v, &si)
})
if err != nil {
si = SiteInfo{Languages: make(map[string]float64), Redirects: make(map[string]string)}
} else {
if si.Languages == nil {
si.Languages = make(map[string]float64)
}
if si.Redirects == nil {
si.Redirects = make(map[string]string)
}
}
d.siteCache[host] = &si
info = &si
}
// 在锁内调用修改函数(直接修改缓存中的对象)
fn(info)
d.siteCacheMu.Unlock()
// 标记脏(锁外)
d.siteDirtyMu.Lock()
d.siteDirty[host] = struct{}{}
d.siteDirtyMu.Unlock()
return nil
}
// ForEachSite 遍历所有网站元信息条目,对每个条目调用 fn。
func (d *DB) ForEachSite(fn func(host string, info *SiteInfo) error) error {
return d.db.View(func(tx *bolt.Tx) error {
return tx.Bucket(bucketSiteGate).ForEach(func(k, v []byte) error {
var info SiteInfo
if err := decompressUnmarshal(v, &info); err != nil {
return nil // 跳过损坏条目
}
return fn(string(k), &info)
})
})
}
// ForEachSnippet 遍历所有 URL 摘要条目,对每个条目调用 fn。
func (d *DB) ForEachSnippet(fn func(url string, entry *SnippetEntry) error) error {
return d.db.View(func(tx *bolt.Tx) error {
return tx.Bucket(bucketGate).ForEach(func(k, v []byte) error {
var entry SnippetEntry
if err := decompressUnmarshal(v, &entry); err != nil {
return nil // 跳过损坏条目
}
return fn(string(k), &entry)
})
})
}
// ---- 优先爬取队列(Priority Queue)相关方法 ----
// PriorityEntry 记录一条待优先爬取的 URL 或域名。
type PriorityEntry struct {
URL string `json:"url"` // 用户提交的 URL 或域名(会自动规范化为带 scheme 的 URL)
IsDomain bool `json:"domain"` // 是否为纯域名(true=仅域名,false=完整 URL
AddedAt int64 `json:"added_at"` // 添加时的 Unix 时间戳
Visited bool `json:"visited"` // 是否已爬取(crawler 爬完后标记)
}
// GetPriorityURLs 返回所有未访问的 priority 条目(按添加时间升序)。
func (d *DB) GetPriorityURLs() ([]PriorityEntry, error) {
var entries []PriorityEntry
err := d.db.View(func(tx *bolt.Tx) error {
return tx.Bucket(bucketPriority).ForEach(func(k, v []byte) error {
var e PriorityEntry
if err := decompressUnmarshal(v, &e); err != nil {
return nil // 跳过损坏条目
}
if !e.Visited {
entries = append(entries, e)
}
return nil
})
})
return entries, err
}
// AddPriorityURL 添加一条 priority 条目(key = URLvalue = PriorityEntry)。
// 每次手动添加都会重新爬取,不做去重检查。
func (d *DB) AddPriorityURL(entry PriorityEntry) error {
return d.db.Update(func(tx *bolt.Tx) error {
data, err := marshalCompress(entry)
if err != nil {
return err
}
return tx.Bucket(bucketPriority).Put([]byte(entry.URL), data)
})
}
// RemovePriorityURL 删除指定 URL 的 priority 条目。
func (d *DB) RemovePriorityURL(url string) error {
return d.db.Update(func(tx *bolt.Tx) error {
return tx.Bucket(bucketPriority).Delete([]byte(url))
})
}
// MarkPriorityURLVisited 将指定 URL 的 priority 条目标记为已访问。
// 用于 priority 爬取完成后标记,避免 RemovePriorityURL 后同一 URL 被重复添加。
func (d *DB) MarkPriorityURLVisited(url string) error {
return d.db.Update(func(tx *bolt.Tx) error {
k := []byte(url)
v := tx.Bucket(bucketPriority).Get(k)
if v == nil {
return nil // 条目不存在,无需处理
}
var e PriorityEntry
if err := decompressUnmarshal(v, &e); err != nil {
return nil
}
if e.Visited {
return nil // 已是 visited 状态
}
e.Visited = true
data, err := marshalCompress(e)
if err != nil {
return err
}
return tx.Bucket(bucketPriority).Put(k, data)
})
}
// ClearVisitedPriorityURLs 批量删除所有已标记为 visited 的条目(crawler 爬完后调用)。
func (d *DB) ClearVisitedPriorityURLs() error {
return d.db.Update(func(tx *bolt.Tx) error {
return tx.Bucket(bucketPriority).ForEach(func(k, v []byte) error {
var e PriorityEntry
if err := decompressUnmarshal(v, &e); err != nil {
return nil
}
if e.Visited {
return tx.Bucket(bucketPriority).Delete(k)
}
return nil
})
})
}
+651
View File
@@ -0,0 +1,651 @@
// Package storage provides unified storage interface backed by Redis + MySQL.
// Redis is used as the primary high-performance store, MySQL is used for persistence.
package storage
import (
"context"
"crypto/md5"
"fmt"
"sync"
"sync/atomic"
goredis "github.com/redis/go-redis/v9"
extredis "sese-engine/redis"
)
// PriorityEntry 记录一条待优先爬取的 URL 或域名。
type PriorityEntry struct {
URL string `json:"url"` // 用户提交的 URL 或域名(会自动规范化为带 scheme 的 URL)
IsDomain bool `json:"domain"` // 是否为纯域名(true=仅域名,false=完整 URL
AddedAt int64 `json:"added_at"` // 添加时的 Unix 时间戳
Visited bool `json:"visited"` // 是否已爬取(crawler 爬完后标记)
}
// Store 统一存储接口,所有数据操作都通过此接口。
type Store interface {
// 倒排索引操作
GetIndex(keyword string) ([]IndexEntry, error)
BatchSetIndex(batch map[string][]IndexEntry) error
ForEachIndex(fn func(keyword string, entries []IndexEntry) error) error
// URL 摘要操作
GetSnippet(url string) (*SnippetEntry, error)
SetSnippet(url string, entry *SnippetEntry) error
ForEachSnippet(fn func(url string, entry *SnippetEntry) error) error
// 网站信息操作
GetSiteInfo(host string) (*SiteInfo, error)
SetSiteInfo(host string, info *SiteInfo) error
UpdateSiteInfo(host string, fn func(*SiteInfo)) error
ForEachSite(fn func(host string, info *SiteInfo) error) error
// Priority URL 操作
GetPriorityURLs() ([]PriorityEntry, error)
AddPriorityURL(entry PriorityEntry) error
RemovePriorityURL(url string) error
MarkPriorityURLVisited(url string) error
ClearVisitedPriorityURLs() error
// 生命周期
Close() error
}
// RedisStore 实现了 Store 接口,提供 Redis 存储能力。
type RedisStoreV2 struct {
client *goredis.Client
// 内存索引聚合(用于写入)
mem map[string][]IndexEntry
memMu sync.RWMutex
rowCount int64
// 站点信息内存缓存
siteCache map[string]*SiteInfo
siteCacheMu sync.RWMutex
}
// NewRedisStoreV2 创建新的 Redis 存储实例
func NewRedisStoreV2() *RedisStoreV2 {
return &RedisStoreV2{
mem: make(map[string][]IndexEntry),
siteCache: make(map[string]*SiteInfo),
}
}
// Init 初始化 Redis 存储
func (r *RedisStoreV2) Init() error {
if extredis.Client == nil {
return fmt.Errorf("redis not initialized, call redis.Open() first")
}
r.client = extredis.Client
return nil
}
// Close 关闭存储
func (r *RedisStoreV2) Close() error {
// 将内存数据刷到 Redis
if err := r.FlushMemToRedis(); err != nil {
return err
}
return nil
}
// ---- 倒排索引操作 ----
// GetIndex 获取关键词的倒排索引
func (r *RedisStoreV2) GetIndex(keyword string) ([]IndexEntry, error) {
ctx := context.Background()
entries, err := r.client.ZRevRangeWithScores(ctx, "idx:"+keyword, 0, -1).Result()
if err != nil {
if err == goredis.Nil {
return nil, nil
}
return nil, err
}
result := make([]IndexEntry, 0, len(entries))
for _, e := range entries {
result = append(result, IndexEntry{
Weight: float32(e.Score),
URL: e.Member.(string),
})
}
return result, nil
}
// BatchSetIndex 批量设置倒排索引
func (r *RedisStoreV2) BatchSetIndex(batch map[string][]IndexEntry) error {
ctx := context.Background()
for keyword, entries := range batch {
if len(entries) == 0 {
continue
}
// 先删除旧的
r.client.Del(ctx, "idx:"+keyword)
// 添加新的
if len(entries) > 0 {
zSlice := make([]goredis.Z, len(entries))
for i, e := range entries {
zSlice[i] = goredis.Z{
Score: float64(e.Weight),
Member: e.URL,
}
}
if err := r.client.ZAdd(ctx, "idx:"+keyword, zSlice...).Err(); err != nil {
return err
}
}
}
return nil
}
// ForEachIndex 遍历所有倒排索引
func (r *RedisStoreV2) ForEachIndex(fn func(keyword string, entries []IndexEntry) error) error {
ctx := context.Background()
var cursor uint64
for {
keys, nextCursor, err := r.client.Scan(ctx, cursor, "idx:*", 1000).Result()
if err != nil {
return err
}
for _, key := range keys {
keyword := key[4:] // 去掉 "idx:" 前缀
entries, err := r.GetIndex(keyword)
if err != nil {
continue
}
if err := fn(keyword, entries); err != nil {
return err
}
}
cursor = nextCursor
if cursor == 0 {
break
}
}
return nil
}
// ---- URL 摘要操作 ----
// GetSnippet 获取 URL 摘要
func (r *RedisStoreV2) GetSnippet(url string) (*SnippetEntry, error) {
ctx := context.Background()
hash := urlHash(url)
data, err := r.client.HGetAll(ctx, "gate:"+hash).Result()
if err != nil {
return nil, err
}
if len(data) == 0 {
return nil, nil
}
return &SnippetEntry{
Title: data["title"],
Description: data["desc"],
Text: data["text"],
Timestamp: parseInt64(data["ts"]),
ContentHash: data["hash"],
}, nil
}
// SetSnippet 设置 URL 摘要
func (r *RedisStoreV2) SetSnippet(url string, entry *SnippetEntry) error {
ctx := context.Background()
hash := urlHash(url)
fields := map[string]interface{}{
"url": url,
"title": entry.Title,
"desc": entry.Description,
"text": entry.Text,
"ts": entry.Timestamp,
"hash": entry.ContentHash,
}
err := r.client.HMSet(ctx, "gate:"+hash, fields).Err()
if err != nil {
return err
}
// 同时存储 URL→hash 的映射
r.client.Set(ctx, "url2hash:"+url, hash, 0)
return nil
}
// ForEachSnippet 遍历所有 URL 摘要
func (r *RedisStoreV2) ForEachSnippet(fn func(url string, entry *SnippetEntry) error) error {
ctx := context.Background()
var cursor uint64
for {
keys, nextCursor, err := r.client.Scan(ctx, cursor, "gate:*", 1000).Result()
if err != nil {
return err
}
for _, key := range keys {
data, err := r.client.HGetAll(ctx, key).Result()
if err != nil || len(data) == 0 {
continue
}
entry := &SnippetEntry{
Title: data["title"],
Description: data["desc"],
Text: data["text"],
Timestamp: parseInt64(data["ts"]),
ContentHash: data["hash"],
}
if err := fn(data["url"], entry); err != nil {
return err
}
}
cursor = nextCursor
if cursor == 0 {
break
}
}
return nil
}
// ---- 网站信息操作 ----
// GetSiteInfo 获取网站信息
func (r *RedisStoreV2) GetSiteInfo(host string) (*SiteInfo, error) {
// 先从内存缓存读取
r.siteCacheMu.RLock()
if info, ok := r.siteCache[host]; ok {
r.siteCacheMu.RUnlock()
return info, nil
}
r.siteCacheMu.RUnlock()
// 从 Redis 读取
ctx := context.Background()
data, err := r.client.HGetAll(ctx, "site:"+host).Result()
if err != nil {
return nil, err
}
if len(data) == 0 {
return nil, nil
}
info := &SiteInfo{
VisitCount: int(parseInt64(data["visit_count"])),
LastVisitTime: parseInt64(data["last_visit_time"]),
}
if v, ok := data["success_rate"]; ok {
f := parseFloat64(v)
info.SuccessRate = &f
}
if v, ok := data["https_available"]; ok {
b := parseInt64(v) == 1
info.HTTPSAvailable = &b
}
// 回填缓存
r.siteCacheMu.Lock()
r.siteCache[host] = info
r.siteCacheMu.Unlock()
return info, nil
}
// SetSiteInfo 设置网站信息
func (r *RedisStoreV2) SetSiteInfo(host string, info *SiteInfo) error {
ctx := context.Background()
fields := map[string]interface{}{
"visit_count": info.VisitCount,
"last_visit_time": info.LastVisitTime,
}
if info.SuccessRate != nil {
fields["success_rate"] = *info.SuccessRate
}
if info.HTTPSAvailable != nil {
if *info.HTTPSAvailable {
fields["https_available"] = 1
} else {
fields["https_available"] = 0
}
}
err := r.client.HMSet(ctx, "site:"+host, fields).Err()
if err != nil {
return err
}
// 更新缓存
r.siteCacheMu.Lock()
r.siteCache[host] = info
r.siteCacheMu.Unlock()
return nil
}
// UpdateSiteInfo 更新网站信息
func (r *RedisStoreV2) UpdateSiteInfo(host string, fn func(*SiteInfo)) error {
info, err := r.GetSiteInfo(host)
if err != nil {
return err
}
if info == nil {
info = &SiteInfo{}
}
// 调用更新函数
fn(info)
return r.SetSiteInfo(host, info)
}
// ForEachSite 遍历所有网站信息
func (r *RedisStoreV2) ForEachSite(fn func(host string, info *SiteInfo) error) error {
ctx := context.Background()
var cursor uint64
for {
keys, nextCursor, err := r.client.Scan(ctx, cursor, "site:*", 1000).Result()
if err != nil {
return err
}
for _, key := range keys {
host := key[5:] // 去掉 "site:" 前缀
info, err := r.GetSiteInfo(host)
if err != nil || info == nil {
continue
}
if err := fn(host, info); err != nil {
return err
}
}
cursor = nextCursor
if cursor == 0 {
break
}
}
return nil
}
// ---- Priority URL 操作 ----
// GetPriorityURLs 获取所有未访问的 Priority URL
func (r *RedisStoreV2) GetPriorityURLs() ([]PriorityEntry, error) {
ctx := context.Background()
keys, err := r.client.Keys(ctx, "priority:*").Result()
if err != nil {
return nil, err
}
var entries []PriorityEntry
for _, key := range keys {
data, err := r.client.HGetAll(ctx, key).Result()
if err != nil || len(data) == 0 {
continue
}
visited := data["visited"] == "1"
if visited {
continue
}
entry := PriorityEntry{
URL: data["url"],
IsDomain: data["is_domain"] == "1",
AddedAt: parseInt64(data["added_at"]),
Visited: visited,
}
entries = append(entries, entry)
}
return entries, nil
}
// AddPriorityURL 添加 Priority URL
func (r *RedisStoreV2) AddPriorityURL(entry PriorityEntry) error {
ctx := context.Background()
fields := map[string]interface{}{
"url": entry.URL,
"is_domain": boolToStr(entry.IsDomain),
"added_at": entry.AddedAt,
"visited": boolToStr(entry.Visited),
}
key := "priority:" + entry.URL
return r.client.HMSet(ctx, key, fields).Err()
}
// RemovePriorityURL 删除 Priority URL
func (r *RedisStoreV2) RemovePriorityURL(url string) error {
ctx := context.Background()
return r.client.Del(ctx, "priority:"+url).Err()
}
// MarkPriorityURLVisited 标记 Priority URL 为已访问
func (r *RedisStoreV2) MarkPriorityURLVisited(url string) error {
ctx := context.Background()
return r.client.HSet(ctx, "priority:"+url, "visited", "1").Err()
}
// ClearVisitedPriorityURLs 清除已访问的 Priority URL
func (r *RedisStoreV2) ClearVisitedPriorityURLs() error {
ctx := context.Background()
keys, err := r.client.Keys(ctx, "priority:*").Result()
if err != nil {
return err
}
for _, key := range keys {
visited, _ := r.client.HGet(ctx, key, "visited").Result()
if visited == "1" {
r.client.Del(ctx, key)
}
}
return nil
}
// ---- 内存索引操作(用于写入)----
// GetMemIndex 获取内存中的索引条目
func (r *RedisStoreV2) GetMemIndex(keyword string) []IndexEntry {
r.memMu.RLock()
defer r.memMu.RUnlock()
return r.mem[keyword]
}
// SetMemIndex 设置内存中的索引条目
func (r *RedisStoreV2) SetMemIndex(keyword string, entries []IndexEntry) {
r.memMu.Lock()
r.mem[keyword] = entries
r.memMu.Unlock()
}
// GetAllMemIndexes 获取所有内存索引
func (r *RedisStoreV2) GetAllMemIndexes() map[string][]IndexEntry {
r.memMu.RLock()
defer r.memMu.RUnlock()
result := make(map[string][]IndexEntry, len(r.mem))
for k, v := range r.mem {
result[k] = v
}
return result
}
// GetRowCount 获取未刷盘的索引条目数
func (r *RedisStoreV2) GetRowCount() int64 {
return atomic.LoadInt64(&r.rowCount)
}
// AddRowCount 增加未刷盘的索引条目计数
func (r *RedisStoreV2) AddRowCount(delta int64) {
atomic.AddInt64(&r.rowCount, delta)
}
// SetRowCount 设置未刷盘的索引条目计数
func (r *RedisStoreV2) SetRowCount(v int64) {
atomic.StoreInt64(&r.rowCount, v)
}
// FlushMemToRedis 将内存索引刷到 Redis
func (r *RedisStoreV2) FlushMemToRedis() error {
r.memMu.Lock()
snapshot := r.mem
r.mem = make(map[string][]IndexEntry)
atomic.StoreInt64(&r.rowCount, 0)
r.memMu.Unlock()
if len(snapshot) == 0 {
return nil
}
// 合并内存和 Redis 数据,然后写回
for keyword, memEntries := range snapshot {
// 从 Redis 读取已有数据
diskEntries, _ := r.GetIndex(keyword)
// 合并
merged := mergeEntries(memEntries, diskEntries)
// 写回 Redis
if err := r.setIndexEntries(keyword, merged); err != nil {
return err
}
}
return nil
}
func (r *RedisStoreV2) setIndexEntries(keyword string, entries []IndexEntry) error {
ctx := context.Background()
// 删除旧的
r.client.Del(ctx, "idx:"+keyword)
if len(entries) > 0 {
zSlice := make([]goredis.Z, len(entries))
for i, e := range entries {
zSlice[i] = goredis.Z{
Score: float64(e.Weight),
Member: e.URL,
}
}
return r.client.ZAdd(ctx, "idx:"+keyword, zSlice...).Err()
}
return nil
}
func mergeEntries(newEntries, existingEntries []IndexEntry) []IndexEntry {
seen := make(map[string]bool)
var result []IndexEntry
// 先添加新条目
for _, e := range newEntries {
if !seen[e.URL] {
result = append(result, e)
seen[e.URL] = true
}
}
// 添加已有条目中不在新条目里的
for _, e := range existingEntries {
if !seen[e.URL] {
result = append(result, e)
seen[e.URL] = true
}
}
return result
}
// SiteCacheRefresh 从 Redis 刷新站点缓存
func (r *RedisStoreV2) SiteCacheRefresh() error {
ctx := context.Background()
var cursor uint64
r.siteCacheMu.Lock()
defer r.siteCacheMu.Unlock()
for {
keys, nextCursor, err := r.client.Scan(ctx, cursor, "site:*", 1000).Result()
if err != nil {
return err
}
for _, key := range keys {
host := key[5:] // 去掉 "site:" 前缀
data, err := r.client.HGetAll(ctx, key).Result()
if err != nil || len(data) == 0 {
continue
}
info := &SiteInfo{
VisitCount: int(parseInt64(data["visit_count"])),
LastVisitTime: parseInt64(data["last_visit_time"]),
}
if v, ok := data["success_rate"]; ok {
f := parseFloat64(v)
info.SuccessRate = &f
}
if v, ok := data["https_available"]; ok {
b := parseInt64(v) == 1
info.HTTPSAvailable = &b
}
r.siteCache[host] = info
}
cursor = nextCursor
if cursor == 0 {
break
}
}
return nil
}
// ---- 辅助函数 ----
func urlHash(url string) string {
return fmt.Sprintf("%x", md5.Sum([]byte(url)))
}
func parseInt64(s string) int64 {
var v int64
fmt.Sscanf(s, "%d", &v)
return v
}
func parseFloat64(s string) float64 {
var v float64
fmt.Sscanf(s, "%f", &v)
return v
}
func boolToStr(b bool) string {
if b {
return "1"
}
return "0"
}
+37
View File
@@ -0,0 +1,37 @@
// Package storage provides unified storage interface backed by Redis + MySQL.
// This file defines the core data types used throughout the storage system.
package storage
// IndexEntry 是倒排索引中的单个条目。
// 一条索引记录表示"某个 URL 与某个关键词的相关性权重"。
type IndexEntry struct {
Weight float32 `json:"w"` // 该 URL 在该关键词下的得分/权重
URL string `json:"u"` // 网页 URL
}
// SnippetEntry 是 URL 对应的摘要信息缓存。
// 包含页面标题、描述、正文片段、抓取时间戳和内容哈希(用于增量重爬检测)。
type SnippetEntry struct {
Title string `json:"title"` // 网页标题
Description string `json:"desc"` // meta description 或自动生成的描述
Text string `json:"text"` // 正文前 N 字符的文本片段
Timestamp int64 `json:"ts"` // 抓取该页面时的 Unix 时间戳
ContentHash string `json:"hash"` // 正文内容的 FNV-1a 哈希(用于增量重爬判断内容是否变化)
}
// SiteInfo 存放每个域名/主机的元信息。
type SiteInfo struct {
VisitCount int `json:"visit_count"` // 累计访问该网站的次数
LastVisitTime int64 `json:"last_visit_time"` // 上次访问该网站的时间戳
Fingerprint any `json:"fingerprint,omitempty"` // 网站指纹(用于识别重复站点)
SuccessRate *float64 `json:"success_rate,omitempty"` // 访问成功率(成功次数/总访问次数)
HTMLStructure string `json:"html_structure,omitempty"` // HTML 结构特征摘要
IPs []string `json:"ips,omitempty"` // 该域名解析出的 IP 列表
Quality *float64 `json:"quality,omitempty"` // 网站质量评分(0~1
HTTPSAvailable *bool `json:"https_available,omitempty"` // 是否支持 HTTPS
Keywords []string `json:"keywords,omitempty"` // 该网站的高频关键词列表
OutLinks []string `json:"out_links,omitempty"` // 从该网站页面提取的出站链接列表
Languages map[string]float64 `json:"languages,omitempty"` // 网站语种分布(语种代码 → 占比)
Redirects map[string]string `json:"redirects,omitempty"` // 重定向链(URL → 最终 URL)
ServerTypes []string `json:"server_types,omitempty"` // 网站使用的 HTTP Server 类型列表
}