实战案例：Go 并发爬虫

场景描述

本案例将带你构建一个高性能、可控制的并发爬虫系统。它使用 Go 语言的 goroutine 实现并发抓取，channel 实现任务分发与结果收集，并集成速率限制（Rate Limiter）以避免被目标网站封禁，最终将结构化数据持久化存储。这个系统是构建更复杂爬虫（如搜索引擎爬虫）的基础。

你将学到

并发模式应用：如何使用 goroutine 池（Worker Pool）模式高效管理大量并发任务。
通信与同步：熟练运用 channel 进行 goroutine 间的数据传递和任务协调，并使用 sync.WaitGroup 进行同步。
速率限制：实现基于 time.Ticker 和 channel 的令牌桶（Token Bucket）算法，控制请求频率。
优雅关停：利用 context.Context 和信号处理，实现爬虫的平滑退出与资源清理。
生产级设计：学习可配置、可扩展的项目结构和错误处理最佳实践。

前置知识

在开始之前，请确保你已掌握以下内容：

架构设计

本爬虫系统采用经典的“生产者-消费者”模型，核心组件如下图所示：

+----------------+      +-----------------+      +----------------+
|  TaskProducer  | ---> |  RateLimiter    | ---> |  WorkerPool    |
| (URL 发送器)    |      | (速率限制器)     |      | (工作者池)      |
+----------------+      +-----------------+      +----------------+
        |                                                    |
        |                                                    |
        v                                                    v
+----------------+                                      +----------------+
|   URL Queue    | <------------------------------------ |  ResultQueue  |
| (任务队列)      |                                      | (结果队列)     |
+----------------+                                      +----------------+
                                                             |
                                                             v
                                                      +----------------+
                                                      | DataPersister  |
                                                      | (数据持久化器)  |
                                                      +----------------+

TaskProducer：读取初始URL，并可能从页面中解析出新的URL，将其发送到 URL Queue。
URL Queue：一个带缓冲的 chan string，连接生产者和消费者。
RateLimiter：一个令牌桶，在发送任务前消耗令牌，确保请求速率不超过设定值。
WorkerPool：一组固定数量的 goroutine，从队列中获取URL，执行抓取和解析。
ResultQueue：一个 chan PageData，工作者将抓取到的数据发送到这里。
DataPersister：从 ResultQueue 读取数据，并写入到文件或数据库。
协调与关停：使用 sync.WaitGroup 等待所有工作者完成，使用 context.Context 传递取消信号。

实现步骤

步骤1：项目初始化与基础结构定义

首先，定义我们爬取的数据结构和一些基础配置。

// main.go
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"golang.org/x/time/rate" // 用于更精确的限速
)

// PageData 代表一个爬取结果
type PageData struct {
	URL     string
	Title   string
	Content string // 简化起见，这里只存字符串
	Error   error
}

// CrawlerConfig 爬虫配置
type CrawlerConfig struct {
	MaxWorkers int
	Rate       rate.Limit // 每秒允许的请求数
	Burst      int        // 令牌桶容量
	OutputFile string
}

步骤2：实现工作者函数（Worker）

这是爬虫的核心执行单元。

// worker 是实际执行抓取任务的函数，运行在单独的goroutine中。
func worker(ctx context.Context, id int, jobs <-chan string, results chan<- PageData, limiter *rate.Limiter, wg *sync.WaitGroup) {
	defer wg.Done()
	client := &http.Client{Timeout: 10 * time.Second}

	for {
		select {
		case <-ctx.Done(): // 收到取消信号
			fmt.Printf("[Worker %d] 接收到停止信号，退出\n", id)
			return
		case url, ok := <-jobs:
			if !ok { // 任务队列已关闭
				fmt.Printf("[Worker %d] 任务队列为空，退出\n", id)
				return
			}
			// 使用限速器
			if err := limiter.Wait(ctx); err != nil {
				fmt.Printf("[Worker %d] 等待限速失败: %v\n", id, err)
				results <- PageData{URL: url, Error: err}
				continue
			}

			fmt.Printf("[Worker %d] 开始抓取: %s\n", id, url)
			data, err := fetchPage(client, url)
			results <- PageData{URL: url, Title: data.Title, Content: data.Content, Error: err}
		}
	}
}

// fetchPage 执行实际的HTTP请求和简单解析
func fetchPage(client *http.Client, url string) (PageData, error) {
	resp, err := client.Get(url)
	if err != nil {
		return PageData{}, err
	}
	defer resp.Body.Close()

	// 这里简化处理，只提取状态码作为示例
	// 实际项目中应使用 golang.org/x/net/html 或 goquery 解析HTML
	return PageData{
		Content: fmt.Sprintf("Status Code: %d", resp.StatusCode),
	}, nil
}

步骤3：创建任务生产者与结果收集器

// startTaskProducer 生成初始任务并可选地从结果中提取新任务（如链接）
func startTaskProducer(ctx context.Context, jobs chan<- string, seedURLs []string) {
	go func() {
		defer close(jobs) // 当所有种子URL都发送完毕后，关闭任务队列
		for _, url := range seedURLs {
			select {
			case <-ctx.Done():
				return
			case jobs <- url:
			}
		}
	}()
}

// startResultCollector 收集结果并执行持久化
func startResultCollector(ctx context.Context, results <-chan PageData, outputFile string, wg *sync.WaitGroup) {
	defer wg.Done()
	// 创建输出文件
	file, err := os.Create(outputFile)
	if err != nil {
		fmt.Printf("创建输出文件失败: %v\n", err)
		return
	}
	defer file.Close()

	for {
		select {
		case <-ctx.Done():
			fmt.Println("[Collector] 接收到停止信号，正在退出")
			return
		case data, ok := <-results:
			if !ok {
				fmt.Println("[Collector] 结果队列已关闭，退出")
				return
			}
			if data.Error != nil {
				fmt.Printf("[Collector] 抓取失败 %s: %v\n", data.URL, data.Error)
				continue
			}
			// 持久化到文件
			_, err := fmt.Fprintf(file, "URL: %s\nContent: %s\n---\n", data.URL, data.Content)
			if err != nil {
				fmt.Printf("[Collector] 写入文件失败: %v\n", err)
			}
			fmt.Printf("[Collector] 成功保存: %s\n", data.URL)
		}
	}
}

步骤4：实现速率限制器

这里我们使用 golang.org/x/time/rate 包中的限速器，它实现了更平滑的令牌桶算法。

// newRateLimiter 创建一个新的限速器实例
func newRateLimiter(config CrawlerConfig) *rate.Limiter {
	// 例如：每秒5个请求，初始桶容量为5个令牌
	return rate.NewLimiter(config.Rate, config.Burst)
}

步骤5：组装主爬虫系统（Main函数）

func main() {
	// 1. 配置
	config := CrawlerConfig{
		MaxWorkers: 5,
		Rate:       rate.Limit(2), // 每秒2个请求
		Burst:      5,
		OutputFile: "results.txt",
	}

	// 2. 创建通道
	jobs := make(chan string, 100)    // 任务队列，带缓冲
	results := make(chan PageData, 100) // 结果队列，带缓冲

	// 3. 创建限速器
	limiter := newRateLimiter(config)

	// 4. 创建上下文（用于优雅关停）
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 5. 设置信号监听，用于优雅关停
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	// 6. 启动工作者池
	var workerWg sync.WaitGroup
	for i := 1; i <= config.MaxWorkers; i++ {
		workerWg.Add(1)
		go worker(ctx, i, jobs, results, limiter, &workerWg)
	}

	// 7. 启动结果收集器
	var collectorWg sync.WaitGroup
	collectorWg.Add(1)
	go startResultCollector(ctx, results, config.OutputFile, &collectorWg)

	// 8. 启动任务生产者
	seedURLs := []string{
		"https://www.example.com",
		"https://golang.org",
		"https://github.com",
	}
	go startTaskProducer(ctx, jobs, seedURLs)

	// 9. 主协程等待：等待任务队列清空，或收到中断信号
	select {
	case <-ctx.Done():
	case <-sigChan:
		fmt.Println("\n[Main] 接收到中断信号，开始优雅关停...")
		cancel() // 触发取消信号
	}

	// 10. 清理工作
	close(jobs)    // 通知工作者停止（虽然生产者也会关闭，但这里再次确保）
	workerWg.Wait() // 等待所有工作者退出
	close(results)  // 关闭结果通道，通知收集器
	collectorWg.Wait() // 等待收集器退出

	fmt.Println("[Main] 爬虫已完全退出。")
}

步骤6：引入依赖并运行

在项目目录下运行：

go mod init go-concurrent-crawler
go get golang.org/x/time/rate
go run main.go

完整项目结构

go-concurrent-crawler/
├── go.mod
├── go.sum
├── main.go           # 主程序入口，组装所有组件
├── worker.go         # 可选：将worker函数抽离到独立文件
├── producer.go       # 可选：将生产者函数抽离
├── collector.go      # 可选：将收集器函数抽离
└── results.txt       # 生成的数据文件

最佳实践

合理设置工作者数量：工作者数量并非越多越好。通常设置为 runtime.NumCPU() 的1-2倍，或根据目标网站的承受能力和网络IO进行压测调整。
错误重试与退避：对于临时性网络错误（如 429 Too Many Requests, 5xx），应实现指数退避（Exponential Backoff）重试机制。
存储优化：对于大规模数据，应使用数据库（如 PostgreSQL, SQLite）或消息队列（如 Kafka）代替简单文件，并实现批量写入（Batch Insert）以提高性能。
遵守 robots.txt：在爬取前，应先解析并遵守目标网站的 robots.txt 协议，这是法律和道德的底线。
监控与日志：添加结构化日志（如使用 zap 或 logrus），记录请求成功/失败数、队列深度、处理速度等关键指标，便于监控和排错。

常见问题

Q1：如何控制爬取速度，避免被封IP？ A1： 本案例中的 RateLimiter 是关键。通过调整 Rate 和 Burst 参数，可以精确控制每秒请求数。此外，可以随机化请求间隔、设置合理的 User-Agent、使用代理IP池等策略。

Q2：如何处理JavaScript渲染的页面？ A2： 本案例使用 net/http 只能获取静态HTML。对于动态页面，需要集成无头浏览器（Headless Browser），如使用 chromedp 或 rod 库。这会显著增加资源消耗，工作者池大小需相应调小。

Q3：如何实现去重，避免重复爬取同一个URL？ A3： 可以使用一个线程安全的布隆过滤器（Bloom Filter）或 map[string]bool（配合 sync.RWMutex）在任务生产者侧进行URL去重。生产者在发送URL前先检查是否已存在。

扩展挑战

分布式爬虫：将任务队列（jobs）从内存中的 channel 替换为 Redis 或 Kafka 等分布式队列，实现跨机器的工作者协作。
深度优先与广度优先策略：修改任务生产者逻辑，实现DFS（使用栈）或BFS（使用队列）来爬取整站链接。
集成代理IP池：实现一个代理IP管理器，在工作者发起请求前，从池中获取一个可用代理，失败则自动切换。

课程	相关章节
Go 编程之旅	Goroutine 基础
Go 编程之旅	并发模式（Channel）
Go 编程之旅	项目结构与最佳实践
Go 编程之旅	错误处理
Go 编程之旅	泛型

Go 并发爬虫