CodeBuddy 协作学习：从小白到熟练掌握 Go 并发编程的成长记录

// 我最初的困惑：Java vs Go 并发模型对比
// Java传统方式
public class JavaConcurrency {
    private ExecutorService executor = Executors.newFixedThreadPool(10);
    
    public void processData(List<String> data) {
        for (String item : data) {
            executor.submit(() -> {
                // 处理数据
                processItem(item);
            });
        }
    }
}

// Go的方式（初学时的疑惑）
func main() {
    data := []string{"item1", "item2", "item3"}
    
    for _, item := range data {
        go processItem(item) // 这样就创建了goroutine？
    }
    
    // 如何等待所有goroutine完成？
    // 如何控制并发数量？
    // 如何处理错误？
}

# Go开发环境配置
# 1. 安装Go语言
go version # go1.21.0

# 2. 配置开发环境
export GOPROXY=https://goproxy.cn,direct
export GOSUMDB=sum.golang.google.cn

# 3. 安装调试和分析工具
go install github.com/go-delve/delve/cmd/dlv@latest
go install golang.org/x/tools/cmd/pprof@latest
go install github.com/google/pprof@latest

# 4. 创建学习项目结构
mkdir go-concurrency-learning
cd go-concurrency-learning
go mod init concurrency-learning

# 项目结构
# ├── basics/          # 基础概念练习
# ├── patterns/        # 并发模式实现
# ├── projects/        # 实战项目
# ├── benchmarks/      # 性能测试
# └── notes/           # 学习笔记

// 第一个goroutine程序 - CodeBuddy指导下的实现
package main

import (
    "fmt"
    "runtime"
    "sync"
    "time"
)

func main() {
    fmt.Printf("开始时的goroutine数量: %d\n", runtime.NumGoroutine())
    
    var wg sync.WaitGroup
    
    // 创建10个goroutine
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Printf("Goroutine %d 开始执行\n", id)
            time.Sleep(time.Millisecond * 100)
            fmt.Printf("Goroutine %d 执行完成\n", id)
        }(i) // 注意：传递参数避免闭包陷阱
    }
    
    fmt.Printf("创建goroutine后的数量: %d\n", runtime.NumGoroutine())
    
    wg.Wait()
    fmt.Printf("所有goroutine完成后的数量: %d\n", runtime.NumGoroutine())
}

// Channel基础使用模式 - 从简单到复杂的学习过程
package main

import (
    "fmt"
    "time"
)

// 1. 无缓冲channel - 同步通信
func unbufferedChannelDemo() {
    ch := make(chan string)
    
    go func() {
        time.Sleep(time.Second)
        ch <- "Hello from goroutine"
    }()
    
    msg := <-ch // 阻塞等待消息
    fmt.Println("接收到消息:", msg)
}

// 2. 有缓冲channel - 异步通信
func bufferedChannelDemo() {
    ch := make(chan int, 3) // 缓冲区大小为3
    
    // 发送方
    go func() {
        for i := 1; i <= 5; i++ {
            ch <- i
            fmt.Printf("发送: %d\n", i)
        }
        close(ch)
    }()
    
    // 接收方
    for num := range ch {
        fmt.Printf("接收: %d\n", num)
        time.Sleep(time.Millisecond * 500)
    }
}

// 3. 双向channel与单向channel
func channelDirectionDemo() {
    ch := make(chan int)
    
    // 只发送channel
    go func(sendCh chan<- int) {
        for i := 0; i < 3; i++ {
            sendCh <- i
        }
        close(sendCh)
    }(ch)
    
    // 只接收channel
    go func(recvCh <-chan int) {
        for num := range recvCh {
            fmt.Printf("处理数据: %d\n", num)
        }
    }(ch)
    
    time.Sleep(time.Second)
}

// Select语句的多种应用场景
package main

import (
    "fmt"
    "time"
)

// 1. 基础select - 多channel选择
func basicSelectDemo() {
    ch1 := make(chan string)
    ch2 := make(chan string)
    
    go func() {
        time.Sleep(time.Millisecond * 100)
        ch1 <- "来自channel1的消息"
    }()
    
    go func() {
        time.Sleep(time.Millisecond * 200)
        ch2 <- "来自channel2的消息"
    }()
    
    select {
    case msg1 := <-ch1:
        fmt.Println("收到:", msg1)
    case msg2 := <-ch2:
        fmt.Println("收到:", msg2)
    case <-time.After(time.Millisecond * 150):
        fmt.Println("超时了")
    }
}

// 2. 非阻塞操作 - default分支
func nonBlockingSelectDemo() {
    ch := make(chan int, 1)
    
    // 非阻塞发送
    select {
    case ch <- 42:
        fmt.Println("成功发送数据")
    default:
        fmt.Println("channel已满，无法发送")
    }
    
    // 非阻塞接收
    select {
    case data := <-ch:
        fmt.Printf("接收到数据: %d\n", data)
    default:
        fmt.Println("channel为空，无法接收")
    }
}

// 3. 心跳检测模式
func heartbeatDemo() {
    heartbeat := time.NewTicker(time.Second)
    timeout := time.NewTimer(time.Second * 5)
    
    defer heartbeat.Stop()
    defer timeout.Stop()
    
    for {
        select {
        case <-heartbeat.C:
            fmt.Println("心跳检测 - 系统正常")
        case <-timeout.C:
            fmt.Println("超时退出")
            return
        }
    }
}

// 1. Worker Pool模式 - 控制并发数量
package main

import (
    "fmt"
    "sync"
    "time"
)

type Job struct {
    ID   int
    Data string
}

type Result struct {
    Job    Job
    Output string
    Error  error
}

func workerPoolDemo() {
    const numWorkers = 3
    const numJobs = 10
    
    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)
    
    // 启动worker goroutines
    var wg sync.WaitGroup
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }
    
    // 发送任务
    go func() {
        for j := 1; j <= numJobs; j++ {
            jobs <- Job{ID: j, Data: fmt.Sprintf("task-%d", j)}
        }
        close(jobs)
    }()
    
    // 等待所有worker完成
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // 收集结果
    for result := range results {
        if result.Error != nil {
            fmt.Printf("任务 %d 失败: %v\n", result.Job.ID, result.Error)
        } else {
            fmt.Printf("任务 %d 完成: %s\n", result.Job.ID, result.Output)
        }
    }
}

func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Worker %d 开始处理任务 %d\n", id, job.ID)
        
        // 模拟工作
        time.Sleep(time.Millisecond * 500)
        
        results <- Result{
            Job:    job,
            Output: fmt.Sprintf("processed by worker %d", id),
            Error:  nil,
        }
    }
}

// Pipeline模式实现 - 数据处理流水线
package main

import (
    "fmt"
    "strconv"
    "strings"
)

// 阶段1：生成数据
func generateNumbers(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// 阶段2：数据转换
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// 阶段3：数据格式化
func toString(in <-chan int) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for n := range in {
            out <- strconv.Itoa(n)
        }
    }()
    return out
}

// 阶段4：数据聚合
func join(in <-chan string, sep string) string {
    var result []string
    for s := range in {
        result = append(result, s)
    }
    return strings.Join(result, sep)
}

func pipelineDemo() {
    // 构建pipeline
    numbers := generateNumbers(1, 2, 3, 4, 5)
    squared := square(numbers)
    strings := toString(squared)
    result := join(strings, ", ")
    
    fmt.Printf("Pipeline结果: %s\n", result)
}

// Fan-out/Fan-in模式实现
package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

// 模拟耗时的工作函数
func expensiveWork(id int, data int) int {
    // 模拟不同的处理时间
    time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
    return data * data
}

// Fan-out: 将工作分发给多个goroutine
func fanOut(input <-chan int, workerCount int) []<-chan int {
    outputs := make([]<-chan int, workerCount)
    
    for i := 0; i < workerCount; i++ {
        output := make(chan int)
        outputs[i] = output
        
        go func(workerID int, out chan<- int) {
            defer close(out)
            for data := range input {
                result := expensiveWork(workerID, data)
                out <- result
            }
        }(i, output)
    }
    
    return outputs
}

// Fan-in: 将多个channel的结果合并到一个channel
func fanIn(inputs ...<-chan int) <-chan int {
    output := make(chan int)
    var wg sync.WaitGroup
    
    // 为每个输入channel启动一个goroutine
    for _, input := range inputs {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for data := range ch {
                output <- data
            }
        }(input)
    }
    
    // 等待所有输入完成后关闭输出channel
    go func() {
        wg.Wait()
        close(output)
    }()
    
    return output
}

func fanOutFanInDemo() {
    // 创建输入数据
    input := make(chan int)
    go func() {
        defer close(input)
        for i := 1; i <= 10; i++ {
            input <- i
        }
    }()
    
    // Fan-out到3个worker
    outputs := fanOut(input, 3)
    
    // Fan-in合并结果
    result := fanIn(outputs...)
    
    // 收集所有结果
    var results []int
    for r := range result {
        results = append(results, r)
    }
    
    fmt.Printf("处理结果: %v\n", results)
}

// Sync包核心组件的实际应用
package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

// 1. Mutex - 互斥锁保护共享资源
type SafeCounter struct {
    mu    sync.Mutex
    count int64
}

func (c *SafeCounter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.count++
}

func (c *SafeCounter) Value() int64 {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count
}

// 2. RWMutex - 读写锁优化读多写少场景
type SafeMap struct {
    mu   sync.RWMutex
    data map[string]int
}

func NewSafeMap() *SafeMap {
    return &SafeMap{
        data: make(map[string]int),
    }
}

func (sm *SafeMap) Set(key string, value int) {
    sm.mu.Lock()
    defer sm.mu.Unlock()
    sm.data[key] = value
}

func (sm *SafeMap) Get(key string) (int, bool) {
    sm.mu.RLock()
    defer sm.mu.RUnlock()
    value, exists := sm.data[key]
    return value, exists
}

// 3. Once - 确保函数只执行一次
var (
    instance *Database
    once     sync.Once
)

type Database struct {
    connection string
}

func GetDatabase() *Database {
    once.Do(func() {
        fmt.Println("初始化数据库连接...")
        time.Sleep(time.Millisecond * 100) // 模拟初始化耗时
        instance = &Database{connection: "connected"}
    })
    return instance
}

// 4. Atomic - 原子操作
type AtomicCounter struct {
    count int64
}

func (c *AtomicCounter) Increment() {
    atomic.AddInt64(&c.count, 1)
}

func (c *AtomicCounter) Value() int64 {
    return atomic.LoadInt64(&c.count)
}

// 性能对比测试
func benchmarkCounters() {
    const iterations = 1000000
    const goroutines = 10
    
    // 测试Mutex版本
    mutexCounter := &SafeCounter{}
    start := time.Now()
    
    var wg sync.WaitGroup
    for i := 0; i < goroutines; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := 0; j < iterations/goroutines; j++ {
                mutexCounter.Increment()
            }
        }()
    }
    wg.Wait()
    
    mutexTime := time.Since(start)
    fmt.Printf("Mutex计数器: %d, 耗时: %v\n", mutexCounter.Value(), mutexTime)
    
    // 测试Atomic版本
    atomicCounter := &AtomicCounter{}
    start = time.Now()
    
    for i := 0; i < goroutines; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := 0; j < iterations/goroutines; j++ {
                atomicCounter.Increment()
            }
        }()
    }
    wg.Wait()
    
    atomicTime := time.Since(start)
    fmt.Printf("Atomic计数器: %d, 耗时: %v\n", atomicCounter.Value(), atomicTime)
    fmt.Printf("性能提升: %.2fx\n", float64(mutexTime)/float64(atomicTime))
}

// Context包的实际应用场景
package main

import (
    "context"
    "fmt"
    "net/http"
    "time"
)

// 1. 超时控制
func timeoutDemo() {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()
    
    result := make(chan string, 1)
    
    go func() {
        // 模拟耗时操作
        time.Sleep(3 * time.Second)
        result <- "操作完成"
    }()
    
    select {
    case res := <-result:
        fmt.Println("结果:", res)
    case <-ctx.Done():
        fmt.Println("操作超时:", ctx.Err())
    }
}

// 2. 取消传播
func cancellationDemo() {
    ctx, cancel := context.WithCancel(context.Background())
    
    // 启动多个goroutine
    for i := 1; i <= 3; i++ {
        go worker(ctx, i)
    }
    
    // 2秒后取消所有操作
    time.Sleep(2 * time.Second)
    fmt.Println("取消所有操作...")
    cancel()
    
    // 等待一段时间观察效果
    time.Sleep(1 * time.Second)
}

func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d 收到取消信号: %v\n", id, ctx.Err())
            return
        default:
            fmt.Printf("Worker %d 正在工作...\n", id)
            time.Sleep(500 * time.Millisecond)
        }
    }
}

// 3. HTTP请求中的Context应用
func httpContextDemo() {
    // 创建带超时的HTTP客户端
    client := &http.Client{
        Timeout: 5 * time.Second,
    }
    
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()
    
    req, err := http.NewRequestWithContext(ctx, "GET", "https://httpbin.org/delay/2", nil)
    if err != nil {
        fmt.Printf("创建请求失败: %v\n", err)
        return
    }
    
    resp, err := client.Do(req)
    if err != nil {
        fmt.Printf("请求失败: %v\n", err)
        return
    }
    defer resp.Body.Close()
    
    fmt.Printf("请求成功，状态码: %d\n", resp.StatusCode)
}

// 4. Context值传递
type contextKey string

const userIDKey contextKey = "userID"

func contextValueDemo() {
    ctx := context.WithValue(context.Background(), userIDKey, "user123")
    
    processRequest(ctx)
}

func processRequest(ctx context.Context) {
    userID := ctx.Value(userIDKey)
    if userID != nil {
        fmt.Printf("处理用户 %s 的请求\n", userID)
    }
    
    // 传递给下一层
    authenticateUser(ctx)
}

func authenticateUser(ctx context.Context) {
    userID := ctx.Value(userIDKey)
    fmt.Printf("验证用户 %s 的身份\n", userID)
}

// Go内存模型实例分析
package main

import (
    "fmt"
    "runtime"
    "sync"
    "time"
)

// 1. 数据竞争示例
var (
    counter int
    mu      sync.Mutex
)

func raceConditionDemo() {
    const goroutines = 1000
    const increments = 1000
    
    var wg sync.WaitGroup
    
    // 不安全的并发访问
    counter = 0
    start := time.Now()
    
    for i := 0; i < goroutines; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := 0; j < increments; j++ {
                counter++ // 数据竞争！
            }
        }()
    }
    
    wg.Wait()
    unsafeTime := time.Since(start)
    unsafeResult := counter
    
    // 安全的并发访问
    counter = 0
    start = time.Now()
    
    for i := 0; i < goroutines; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := 0; j < increments; j++ {
                mu.Lock()
                counter++
                mu.Unlock()
            }
        }()
    }
    
    wg.Wait()
    safeTime := time.Since(start)
    safeResult := counter
    
    fmt.Printf("不安全版本: 结果=%d (期望=%d), 耗时=%v\n", 
        unsafeResult, goroutines*increments, unsafeTime)
    fmt.Printf("安全版本: 结果=%d (期望=%d), 耗时=%v\n", 
        safeResult, goroutines*increments, safeTime)
}

// 2. Happens-Before关系示例
func happensBefore() {
    var a, b int
    var wg sync.WaitGroup
    
    wg.Add(2)
    
    // Goroutine 1
    go func() {
        defer wg.Done()
        a = 1
        b = 2
    }()
    
    // Goroutine 2
    go func() {
        defer wg.Done()
        fmt.Printf("a=%d, b=%d\n", a, b)
    }()
    
    wg.Wait()
}

// 3. 使用channel建立同步关系
func channelSynchronization() {
    var a, b int
    done := make(chan bool)
    
    // Goroutine 1
    go func() {
        a = 1
        b = 2
        done <- true // 发送信号
    }()
    
    // Goroutine 2
    go func() {
        <-done // 等待信号
        fmt.Printf("a=%d, b=%d\n", a, b) // 保证能看到正确的值
    }()
    
    time.Sleep(time.Millisecond * 100)
}

// 高并发Web服务器实现
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "runtime"
    "sync"
    "time"
)

// 请求统计
type Stats struct {
    mu           sync.RWMutex
    totalReqs    int64
    activeReqs   int64
    avgResponse  time.Duration
    responses    []time.Duration
}

func (s *Stats) RecordRequest(duration time.Duration) {
    s.mu.Lock()
    defer s.mu.Unlock()
    
    s.totalReqs++
    s.responses = append(s.responses, duration)
    
    // 保持最近1000个响应时间用于计算平均值
    if len(s.responses) > 1000 {
        s.responses = s.responses[1:]
    }
    
    // 计算平均响应时间
    var total time.Duration
    for _, d := range s.responses {
        total += d
    }
    s.avgResponse = total / time.Duration(len(s.responses))
}

func (s *Stats) IncrementActive() {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.activeReqs++
}

func (s *Stats) DecrementActive() {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.activeReqs--
}

func (s *Stats) GetStats() (int64, int64, time.Duration) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    return s.totalReqs, s.activeReqs, s.avgResponse
}

var globalStats = &Stats{}

// 中间件：请求统计
func statsMiddleware(next http.HandlerFunc) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        start := time.Now()
        globalStats.IncrementActive()
        defer func() {
            globalStats.DecrementActive()
            globalStats.RecordRequest(time.Since(start))
        }()
        
        next(w, r)
    }
}

// 中间件：超时控制
func timeoutMiddleware(timeout time.Duration) func(http.HandlerFunc) http.HandlerFunc {
    return func(next http.HandlerFunc) http.HandlerFunc {
        return func(w http.ResponseWriter, r *http.Request) {
            ctx, cancel := context.WithTimeout(r.Context(), timeout)
            defer cancel()
            
            r = r.WithContext(ctx)
            
            done := make(chan struct{})
            go func() {
                defer close(done)
                next(w, r)
            }()
            
            select {
            case <-done:
                // 请求正常完成
            case <-ctx.Done():
                http.Error(w, "Request timeout", http.StatusRequestTimeout)
            }
        }
    }
}

// 业务处理器
func heavyWorkHandler(w http.ResponseWriter, r *http.Request) {
    // 模拟CPU密集型工作
    start := time.Now()
    
    // 使用goroutine池处理并发任务
    const workers = 4
    const tasks = 100
    
    jobs := make(chan int, tasks)
    results := make(chan int, tasks)
    
    // 启动worker
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range jobs {
                // 检查context是否被取消
                select {
                case <-r.Context().Done():
                    return
                default:
                    // 模拟计算工作
                    result := job * job
                    results <- result
                }
            }
        }()
    }
    
    // 发送任务
    go func() {
        defer close(jobs)
        for i := 1; i <= tasks; i++ {
            select {
            case jobs <- i:
            case <-r.Context().Done():
                return
            }
        }
    }()
    
    // 等待完成并关闭结果channel
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // 收集结果
    var sum int
    for result := range results {
        sum += result
    }
    
    response := map[string]interface{}{
        "result":    sum,
        "duration":  time.Since(start).String(),
        "goroutines": runtime.NumGoroutine(),
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(response)
}

// 统计信息处理器
func statsHandler(w http.ResponseWriter, r *http.Request) {
    total, active, avg := globalStats.GetStats()
    
    stats := map[string]interface{}{
        "total_requests":    total,
        "active_requests":   active,
        "avg_response_time": avg.String(),
        "goroutines":       runtime.NumGoroutine(),
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(stats)
}

func main() {
    // 设置最大CPU核心数
    runtime.GOMAXPROCS(runtime.NumCPU())
    
    mux := http.NewServeMux()
    
    // 注册路由和中间件
    mux.HandleFunc("/work", 
        statsMiddleware(
            timeoutMiddleware(5*time.Second)(heavyWorkHandler)))
    
    mux.HandleFunc("/stats", statsHandler)
    
    server := &http.Server{
        Addr:         ":8080",
        Handler:      mux,
        ReadTimeout:  10 * time.Second,
        WriteTimeout: 10 * time.Second,
        IdleTimeout:  60 * time.Second,
    }
    
    fmt.Println("服务器启动在 :8080")
    log.Fatal(server.ListenAndServe())
}

// 分布式爬虫系统核心组件
package main

import (
    "context"
    "fmt"
    "net/http"
    "sync"
    "time"
)

// URL任务
type Task struct {
    URL   string
    Depth int
}

// 爬取结果
type Result struct {
    URL     string
    Content string
    Links   []string
    Error   error
}

// 爬虫配置
type CrawlerConfig struct {
    MaxWorkers   int
    MaxDepth     int
    RequestDelay time.Duration
    Timeout      time.Duration
}

// 分布式爬虫
type DistributedCrawler struct {
    config    CrawlerConfig
    client    *http.Client
    visited   sync.Map
    taskQueue chan Task
    results   chan Result
    ctx       context.Context
    cancel    context.CancelFunc
    wg        sync.WaitGroup
}

func NewCrawler(config CrawlerConfig) *DistributedCrawler {
    ctx, cancel := context.WithCancel(context.Background())
    
    return &DistributedCrawler{
        config: config,
        client: &http.Client{
            Timeout: config.Timeout,
        },
        taskQueue: make(chan Task, config.MaxWorkers*2),
        results:   make(chan Result, config.MaxWorkers*2),
        ctx:       ctx,
        cancel:    cancel,
    }
}

// 启动爬虫
func (c *DistributedCrawler) Start() {
    // 启动worker goroutines
    for i := 0; i < c.config.MaxWorkers; i++ {
        c.wg.Add(1)
        go c.worker(i)
    }
    
    // 启动结果处理器
    go c.resultProcessor()
}

// Worker goroutine
func (c *DistributedCrawler) worker(id int) {
    defer c.wg.Done()
    
    for {
        select {
        case task := <-c.taskQueue:
            fmt.Printf("Worker %d 处理: %s (深度: %d)\n", id, task.URL, task.Depth)
            
            // 检查是否已访问
            if _, loaded := c.visited.LoadOrStore(task.URL, true); loaded {
                continue
            }
            
            // 执行爬取
            result := c.crawlURL(task)
            
            select {
            case c.results <- result:
            case <-c.ctx.Done():
                return
            }
            
            // 添加新发现的链接到任务队列
            if task.Depth < c.config.MaxDepth && result.Error == nil {
                for _, link := range result.Links {
                    newTask := Task{
                        URL:   link,
                        Depth: task.Depth + 1,
                    }
                    
                    select {
                    case c.taskQueue <- newTask:
                    case <-c.ctx.Done():
                        return
                    default:
                        // 队列满了，跳过这个链接
                    }
                }
            }
            
            // 请求间隔
            time.Sleep(c.config.RequestDelay)
            
        case <-c.ctx.Done():
            return
        }
    }
}

// 爬取单个URL
func (c *DistributedCrawler) crawlURL(task Task) Result {
    req, err := http.NewRequestWithContext(c.ctx, "GET", task.URL, nil)
    if err != nil {
        return Result{URL: task.URL, Error: err}
    }
    
    resp, err := c.client.Do(req)
    if err != nil {
        return Result{URL: task.URL, Error: err}
    }
    defer resp.Body.Close()
    
    // 这里应该解析HTML并提取链接
    // 为简化示例，我们模拟返回一些链接
    links := []string{
        task.URL + "/page1",
        task.URL + "/page2",
    }
    
    return Result{
        URL:     task.URL,
        Content: fmt.Sprintf("Content from %s", task.URL),
        Links:   links,
        Error:   nil,
    }
}

// 结果处理器
func (c *DistributedCrawler) resultProcessor() {
    for {
        select {
        case result := <-c.results:
            if result.Error != nil {
                fmt.Printf("爬取失败 %s: %v\n", result.URL, result.Error)
            } else {
                fmt.Printf("爬取成功 %s: %d 个链接\n", result.URL, len(result.Links))
            }
        case <-c.ctx.Done():
            return
        }
    }
}

// 添加种子URL
func (c *DistributedCrawler) AddSeed(url string) {
    task := Task{URL: url, Depth: 0}
    select {
    case c.taskQueue <- task:
    case <-c.ctx.Done():
    }
}

// 停止爬虫
func (c *DistributedCrawler) Stop() {
    c.cancel()
    c.wg.Wait()
    close(c.taskQueue)
    close(c.results)
}

// 使用示例
func main() {
    config := CrawlerConfig{
        MaxWorkers:   5,
        MaxDepth:     2,
        RequestDelay: time.Millisecond * 100,
        Timeout:      time.Second * 10,
    }
    
    crawler := NewCrawler(config)
    crawler.Start()
    
    // 添加种子URL
    crawler.AddSeed("https://example.com")
    
    // 运行一段时间
    time.Sleep(time.Second * 30)
    
    crawler.Stop()
    fmt.Println("爬虫已停止")
}

// 性能分析和优化示例
package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    _ "net/http/pprof"
    "runtime"
    "sync"
    "time"
)

// 性能监控器
type PerformanceMonitor struct {
    mu          sync.RWMutex
    metrics     map[string]time.Duration
    counters    map[string]int64
    startTimes  map[string]time.Time
}

func NewPerformanceMonitor() *PerformanceMonitor {
    return &PerformanceMonitor{
        metrics:    make(map[string]time.Duration),
        counters:   make(map[string]int64),
        startTimes: make(map[string]time.Time),
    }
}

func (pm *PerformanceMonitor) StartTimer(name string) {
    pm.mu.Lock()
    defer pm.mu.Unlock()
    pm.startTimes[name] = time.Now()
}

func (pm *PerformanceMonitor) EndTimer(name string) {
    pm.mu.Lock()
    defer pm.mu.Unlock()
    
    if start, exists := pm.startTimes[name]; exists {
        duration := time.Since(start)
        pm.metrics[name] = duration
        delete(pm.startTimes, name)
    }
}

func (pm *PerformanceMonitor) IncrementCounter(name string) {
    pm.mu.Lock()
    defer pm.mu.Unlock()
    pm.counters[name]++
}

func (pm *PerformanceMonitor) GetMetrics() map[string]interface{} {
    pm.mu.RLock()
    defer pm.mu.RUnlock()
    
    result := make(map[string]interface{})
    
    // 复制metrics
    for k, v := range pm.metrics {
        result[k] = v.String()
    }
    
    // 复制counters
    for k, v := range pm.counters {
        result[k] = v
    }
    
    // 添加运行时信息
    var m runtime.MemStats
    runtime.ReadMemStats(&m)
    
    result["goroutines"] = runtime.NumGoroutine()
    result["heap_alloc"] = m.HeapAlloc
    result["heap_sys"] = m.HeapSys
    result["gc_cycles"] = m.NumGC
    
    return result
}

var monitor = NewPerformanceMonitor()

// 优化前的版本 - 存在性能问题
func inefficientHandler(w http.ResponseWriter, r *http.Request) {
    monitor.StartTimer("inefficient_request")
    defer monitor.EndTimer("inefficient_request")
    monitor.IncrementCounter("inefficient_requests")
    
    // 问题1: 每次都创建新的goroutine，没有复用
    var wg sync.WaitGroup
    results := make(chan int, 100)
    
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            // 问题2: 不必要的内存分配
            data := make([]int, 1000)
            for j := range data {
                data[j] = n * j
            }
            
            sum := 0
            for _, v := range data {
                sum += v
            }
            results <- sum
        }(i)
    }
    
    go func() {
        wg.Wait()
        close(results)
    }()
    
    total := 0
    for result := range results {
        total += result
    }
    
    fmt.Fprintf(w, "结果: %d", total)
}

// 优化后的版本
type WorkerPool struct {
    workers chan chan int
    jobs    chan int
    results chan int
    ctx     context.Context
    cancel  context.CancelFunc
    wg      sync.WaitGroup
}

func NewWorkerPool(numWorkers int) *WorkerPool {
    ctx, cancel := context.WithCancel(context.Background())
    
    wp := &WorkerPool{
        workers: make(chan chan int, numWorkers),
        jobs:    make(chan int, 100),
        results: make(chan int, 100),
        ctx:     ctx,
        cancel:  cancel,
    }
    
    // 启动workers
    for i := 0; i < numWorkers; i++ {
        wp.wg.Add(1)
        go wp.worker()
    }
    
    return wp
}

func (wp *WorkerPool) worker() {
    defer wp.wg.Done()
    
    jobChan := make(chan int)
    
    for {
        select {
        case wp.workers <- jobChan:
            select {
            case job := <-jobChan:
                // 优化: 复用计算逻辑，减少内存分配
                result := wp.processJob(job)
                wp.results <- result
            case <-wp.ctx.Done():
                return
            }
        case <-wp.ctx.Done():
            return
        }
    }
}

func (wp *WorkerPool) processJob(n int) int {
    // 优化: 直接计算，避免不必要的内存分配
    sum := 0
    for j := 0; j < 1000; j++ {
        sum += n * j
    }
    return sum
}

func (wp *WorkerPool) Submit(job int) {
    select {
    case wp.jobs <- job:
    case <-wp.ctx.Done():
    }
}

func (wp *WorkerPool) GetResult() int {
    select {
    case result := <-wp.results:
        return result
    case <-wp.ctx.Done():
        return 0
    }
}

func (wp *WorkerPool) Stop() {
    wp.cancel()
    wp.wg.Wait()
}

var workerPool = NewWorkerPool(10)

func efficientHandler(w http.ResponseWriter, r *http.Request) {
    monitor.StartTimer("efficient_request")
    defer monitor.EndTimer("efficient_request")
    monitor.IncrementCounter("efficient_requests")
    
    // 提交任务到worker pool
    for i := 0; i < 100; i++ {
        workerPool.Submit(i)
    }
    
    // 收集结果
    total := 0
    for i := 0; i < 100; i++ {
        total += workerPool.GetResult()
    }
    
    fmt.Fprintf(w, "结果: %d", total)
}

// 性能监控端点
func metricsHandler(w http.ResponseWriter, r *http.Request) {
    metrics := monitor.GetMetrics()
    
    w.Header().Set("Content-Type", "application/json")
    fmt.Fprintf(w, "{\n")
    for k, v := range metrics {
        fmt.Fprintf(w, "  \"%s\": \"%v\",\n", k, v)
    }
    fmt.Fprintf(w, "}\n")
}

func main() {
    // 启动pprof服务器
    go func() {
        log.Println("pprof服务器启动在 :6060")
        log.Println(http.ListenAndServe(":6060", nil))
    }()
    
    http.HandleFunc("/inefficient", inefficientHandler)
    http.HandleFunc("/efficient", efficientHandler)
    http.HandleFunc("/metrics", metricsHandler)
    
    fmt.Println("服务器启动在 :8080")
    fmt.Println("性能分析: http://localhost:6060/debug/pprof/")
    fmt.Println("指标监控: http://localhost:8080/metrics")
    
    log.Fatal(http.ListenAndServe(":8080", nil))
}

// 常见陷阱与解决方案总结
package main

import (
    "fmt"
    "sync"
    "time"
)

// 陷阱1: Goroutine泄漏
func goroutineLeakExample() {
    // 错误做法 - goroutine永远不会结束
    ch := make(chan int)
    go func() {
        for {
            select {
            case data := <-ch:
                fmt.Println("处理数据:", data)
            // 缺少退出条件！
            }
        }
    }()
    
    // 正确做法 - 提供退出机制
    done := make(chan struct{})
    go func() {
        for {
            select {
            case data := <-ch:
                fmt.Println("处理数据:", data)
            case <-done:
                fmt.Println("Goroutine退出")
                return
            }
        }
    }()
    
    // 记得关闭done channel
    defer close(done)
}

// 陷阱2: Channel死锁
func channelDeadlockExample() {
    // 错误做法 - 无缓冲channel的死锁
    ch := make(chan int)
    
    // 这会导致死锁，因为没有接收者
    // ch <- 1
    
    // 正确做法1 - 使用缓冲channel
    bufferedCh := make(chan int, 1)
    bufferedCh <- 1
    fmt.Println("发送成功:", <-bufferedCh)
    
    // 正确做法2 - 在goroutine中发送
    unbufferedCh := make(chan int)
    go func() {
        unbufferedCh <- 1
    }()
    fmt.Println("接收到:", <-unbufferedCh)
}

// 陷阱3: 循环变量闭包问题
func closureTrapExample() {
    var wg sync.WaitGroup
    
    // 错误做法 - 所有goroutine都会打印最后一个值
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            fmt.Printf("错误: %d\n", i) // 总是打印5
        }()
    }
    
    // 正确做法 - 传递参数
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            fmt.Printf("正确: %d\n", n)
        }(i)
    }
    
    wg.Wait()
}

// 陷阱4: 忘记关闭channel
func channelCloseExample() {
    ch := make(chan int, 10)
    
    // 生产者
    go func() {
        for i := 0; i < 5; i++ {
            ch <- i
        }
        // 重要：记得关闭channel
        close(ch)
    }()
    
    // 消费者 - 使用range自动处理channel关闭
    for value := range ch {
        fmt.Println("接收:", value)
    }
}