Go并发爬虫：解决通道死锁导致的停滞问题_石家庄创泽智能科技有限公司

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

Go并发爬虫：解决通道死锁导致的停滞问题

2025-12-05

浏览次数：次

返回列表

Go并发爬虫：解决通道死锁导致的停滞问题

本文深入探讨了go语言并发爬虫在处理失败url重入队列时可能遇到的通道死锁问题。通过分析原始设计中所有工作协程同时阻塞在输入通道的缺陷，提出了引入独立“失败”通道的解决方案。文章提供了详细的代码示例，并解析了如何通过`select`语句高效管理任务分发与失败重试，确保爬虫稳定运行，避免因并发逻辑不当导致的程序停滞。

Go并发爬虫中的通道死锁问题

Go语言以其强大的并发特性和Goroutine、Channel机制，成为构建高性能并发爬虫的理想选择。然而，在设计复杂的任务调度和错误重试逻辑时，如果不慎处理通道间的交互，很容易引入死锁，导致程序意外停滞。

一个典型的Go并发爬虫结构通常包括：

任务输入通道 (input channel)：用于向工作协程分发待处理的URL。
结果输出通道 (output channel)：用于收集工作协程处理完成的数据。
工作协程 (worker goroutines)：从输入通道接收URL，执行下载、处理等任务，并将结果发送到输出通道。
调度器/协调器 (coordinator goroutine)：负责初始化工作协程，将初始URL推入输入通道，并从输出通道接收并保存结果。

问题现象：爬虫停滞不前

在某些爬虫实现中，为了确保所有URL都能被成功处理，会设计一个重试机制：如果一个URL在处理过程中失败（例如，HTTP请求失败），它会被重新放回输入通道，等待再次处理。这种设计在理论上看似合理，但在高并发场景下，尤其是在所有工作协程同时遇到失败并尝试重入队列时，可能导致程序在运行一段时间后无故停滞。

用户反馈的现象是，爬虫在运行几分钟后（例如5-10分钟）便“卡住”，即使待处理的URL列表尚未耗尽，也无法继续工作。经过排查，并非目标网站的封禁，也不是数据库写入问题，而是程序内部的并发逻辑出现了问题。

死锁根源：重入队列的机制缺陷

导致这种停滞的根本原因在于通道死锁。让我们分析一下原始的worker和crawl函数片段：

func worker(input chan string, output chan SiteData) {
    for url := range input { // (A) 从输入通道接收URL
        resp, status := downloadURL(url)
        if resp != nil && status == 200 {
            output <- processSiteData(resp)
        } else {
            input <- url // (B) 失败时将URL重新放回输入通道
        }
    }
}

func crawl(urlList []string) {
    numWorkers := 4
    input := make(chan string)
    output := make(chan SiteData)

    for i := 0; i < numWorkers; i++ {
        go worker(input, output)
    }

    go func() { // (C) 初始URL分发协程
        for url := range urlList {
            input <- url
        }
    }()

    for { // (D) 结果收集协程
        select {
        case data := <-output:
            s*eToDB(data)
        }
    }
}

死锁场景分析：

假设input通道是无缓冲的（make(chan string)）。
当所有numWorkers个工作协程（例如4个）在处理URL时，都遇到了失败情况。
这4个工作协程会同时尝试执行 input
由于input通道是无缓冲的，并且当前没有其他协程从input通道读取数据，这4个发送操作将全部阻塞。
初始URL分发协程（点C）在将所有初始URL发送完毕后，会因为for url := range urlList循环结束而退出（或者，如果urlList很大，它也会在某个时刻尝试向一个已满的input通道发送数据而阻塞）。
结果收集协程（点D）只负责从output通道接收数据，它不会与input通道交互。

最终结果是：所有工作协程都阻塞在向input通道发送数据，而没有协程从input通道接收数据，从而形成一个典型的发送-发送死锁。程序中的所有活动协程都处于阻塞状态，导致整个程序停滞。

解决方案：引入独立的失败通道

为了解决上述死锁问题，核心思想是将失败任务的重入逻辑与正常任务的分发逻辑解耦。我们可以引入一个独立的“失败通道” (failed chan string) 来专门收集那些需要重试的URL。

设计思路：分离成功与失败任务流

工作协程不再直接向input通道重发失败URL，而是将失败的URL发送到failed通道。
调度器协程需要同时监听初始URL列表的分发和failed通道的重试请求，将它们统一管理到待处理URL列表中，并适时地将URL推送到input通道供工作协程处理。

重构 worker 函数

worker函数现在需要接收三个通道：input、output和failed。

Lateral App

整理归类论文

85 查看详情 Lateral App

func worker(input chan string, output chan SiteData, failed chan string) {
    for url := range input {
        resp, status := downloadURL(url)
        if resp != nil && status == 200 {
            output <- processSiteData(resp)
        } else {
            failed <- url // 将失败的URL发送到独立的failed通道
        }
    }
}

重构 crawl 函数与任务调度器

crawl函数中的任务调度逻辑将变得更加复杂，它需要一个中心化的协程来管理URL列表，并使用select语句来非阻塞地处理新的URL分发和失败URL的重试。

func crawl(urlList []string) {
    numWorkers := 4
    input := make(chan string)
    failed := make(chan string)
    output := make(chan SiteData)

    // 1. 启动工作协程
    for i := 0; i < numWorkers; i++ {
        go worker(input, output, failed)
    }

    // 2. 任务调度协程：负责分发URL和处理失败重试
    go func() {
        pendingURLs := urlList // 维护一个动态的待处理URL列表
        for {
            // 如果没有待处理的URL，则等待失败的URL或退出
            if len(pendingURLs) == 0 {
                select {
                case url := <-failed: // 仅接收失败的URL
                    pendingURLs = append(pendingURLs, url)
                // TODO: 添加一个退出机制，当所有任务完成时关闭通道
                }
            } else {
                // 使用select同时尝试发送URL到input通道和接收失败URL
                select {
                case input <- pendingURLs[0]: // 尝试发送第一个待处理URL
                    pendingURLs = pendingURLs[1:] // 发送成功则移除
                case url := <-failed: // 接收失败的URL并重新加入列表
                    pendingURLs = append(pendingURLs, url)
                }
            }
            // 考虑添加一个退出条件，例如当pendingURLs为空且所有worker都已完成时
        }
    }()

    // 3. 结果收集协程
    for {
        data := <-output
        s*eToDB(data)
        // TODO: 添加一个退出机制，当所有任务完成时关闭通道
    }
}

完整代码示例与详细解析

为了使crawl函数能够优雅地退出，我们需要更精细地管理pendingURLs列表以及判断何时所有任务都已完成。以下是一个更完善的示例：

package main

import (
    "bytes"
    "fmt"
    "io/ioutil"
    "net/http"
    "sync"
    "time"
)

// SiteData 模拟网站数据结构
type SiteData struct {
    URL    string
    Status int
    BodyLen int
    // ... 其他处理后的数据
}

// downloadURL 模拟下载URL内容
func downloadURL(url string) (body []byte, status int) {
    fmt.Printf("Downloading: %s\n", url)
    resp, err := http.Get(url)
    if err != nil {
        fmt.Printf("Error downloading %s: %v\n", url, err)
        return nil, 0
    }
    defer resp.Body.Close()

    status = resp.StatusCode
    if status != 200 {
        fmt.Printf("Non-200 status for %s: %d\n", url, status)
        return nil, status
    }

    body, err = ioutil.ReadAll(resp.Body)
    if err != nil {
        fmt.Printf("Error reading body for %s: %v\n", url, err)
        return nil, status
    }
    body = bytes.Trim(body, "\x00") // 移除可能的空字节

    // 模拟随机失败
    if url == "http://example.com/fail1" || url == "http://example.com/fail2" {
        fmt.Printf("Simulating failure for %s\n", url)
        return nil, 500 // 模拟失败
    }

    time.Sleep(50 * time.Millisecond) // 模拟下载耗时
    return body, status
}

// processSiteData 模拟数据处理
func processSiteData(url string, resp []byte) SiteData {
    fmt.Printf("Processing: %s (body len: %d)\n", url, len(resp))
    time.Sleep(20 * time.Millisecond) // 模拟处理耗时
    return SiteData{URL: url, Status: 200, BodyLen: len(resp)}
}

// s*eToDB 模拟数据保存到数据库
func s*eToDB(data SiteData) {
    fmt.Printf("S*ing to DB: %s (Status: %d, BodyLen: %d)\n", data.URL, data.Status, data.BodyLen)
    time.Sleep(10 * time.Millisecond) // 模拟DB写入耗时
}

// worker 协程：从input接收URL，处理后发送到output或failed
func worker(id int, input chan string, output chan SiteData, failed chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    for url := range input {
        body, status := downloadURL(url)

        if body != nil && status == 200 {
            output <- processSiteData(url, body)
        } else {
            fmt.Printf("Worker %d: URL %s failed, re-enqueuing.\n", id, url)
            failed <- url
        }
    }
    fmt.Printf("Worker %d finished.\n", id)
}

// crawl 主调度函数
func crawl(initialURLs []string) {
    numWorkers := 4
    input := make(chan string)
    failed := make(chan string)
    output := make(chan SiteData)
    done := make(chan struct{}) // 用于通知所有任务完成

    var wg sync.WaitGroup // 用于等待所有worker协程完成

    // 1. 启动工作协程
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go worker(i+1, input, output, failed, &wg)
    }

    // 2. 任务调度协程：负责分发URL和处理失败重试
    go func() {
        pendingURLs := make([]string, len(initialURLs))
        copy(pendingURLs, initialURLs) // 复制初始URL列表

        processedCount := 0
        totalTasks := len(initialURLs) // 初始任务数

        // 用于跟踪当前正在处理的任务数，以便判断何时所有任务完成
        // 这里的逻辑需要更严谨，实际应该通过计数器追踪
        // 为简化示例，假设当pendingURLs为空且没有新的失败任务时，所有任务完成
        // 真正的完成判断需要考虑所有worker是否都已空闲
        // 这里我们使用一个简单的计数器来模拟完成
        var activeTasks int32 // 活跃任务数，包括正在处理和待处理的

        // 初始化活跃任务数
        for _, url := range initialURLs {
            input <- url // 初始分发，这里是阻塞的，如果input无缓冲，可能需要调整
            activeTasks++
        }
        close(input) // 初始URL分发完毕，关闭input通道，让worker知道何时停止

        // 改进的调度器，使用一个单独的协程来管理URL列表
        // 这样可以避免在主调度器中阻塞
        go func() {
            var currentURLs []string
            currentURLs = append(currentURLs, initialURLs...)

            // 确保所有初始URL都已发送到input
            for _, url := range initialURLs {
                input <- url
            }

            // 跟踪正在处理的URL数量
            inFlight := 0

            for {
                select {
                case url := <-failed: // 接收失败的URL
                    currentURLs = append(currentURLs, url)
                    inFlight-- // 失败任务不再in-flight
                    fmt.Printf("Scheduler: Received failed URL: %s, currentURLs len: %d, inFlight: %d\n", url, len(currentURLs), inFlight)
                case input <- currentURLs[0]: // 尝试发送下一个URL
                    fmt.Printf("Scheduler: Sending URL: %s, currentURLs len: %d, inFlight: %d\n", currentURLs[0], len(currentURLs), inFlight)
                    currentURLs = currentURLs[1:]
                    inFlight++ // 成功发送，in-flight任务增加
                    if len(currentURLs) == 0 && inFlight == 0 {
                        // 所有URL都已处理完毕，且没有正在进行中的任务
                        close(input) // 关闭input通道，通知worker停止
                        return
                    }
                }
            }
        }()

        // 这是一个简化的调度器，更健壮的调度器需要更复杂的逻辑
        // 实际应用中，需要一个机制来判断何时所有URL都已成功处理或重试次数耗尽
        // 并且所有的worker都已完成。这里为了避免死锁，我们采用如下策略：
        // 初始URL一次性发送，failed通道接收的URL会重新进入队列。
        // 当input通道关闭后，worker会退出。

        // 这里需要一个更精细的调度器，来动态管理 `input` 和 `failed`
        // 让我们重写这部分，以避免死锁并允许优雅退出
        go func() {
            var urlsToProcess []string
            urlsToProcess = append(urlsToProcess, initialURLs...)

            // 用于在没有URL可发送时，等待失败URL
            sendOrReceive := func() {
                if len(urlsToProcess) > 0 {
                    select {
                    case input <- urlsToProcess[0]:
                        urlsToProcess = urlsToProcess[1:]
                    case url := <-failed:
                        urlsToProcess = append(urlsToProcess, url)
                    }
                } else {
                    // 如果没有待处理URL，则只监听failed通道
                    // 这里是关键：防止在没有URL时阻塞在input <-
                    url := <-failed
                    urlsToProcess = append(urlsToProcess, url)
                }
            }

            // 持续调度，直到所有任务完成
            // 这里需要一个更精细的WaitGroup来跟踪所有任务的状态
            // 为了避免死锁，我们暂时让这个调度器一直运行
            // 直到main函数通过done通道通知其退出
            // 这是一个简化的版本，实际需要一个计数器来跟踪in-flight任务
            for {
                select {
                case <-done: // 收到退出信号
                    close(input) // 关闭input通道，通知worker退出
                    return
                default:
                    sendOrReceive()
                }
            }
        }()
    }()

    // 3. 结果收集协程
    go func() {
        totalResults := 0
        for range output { // 接收所有结果
            totalResults++
            // s*eToDB(data) // 已经在worker中模拟保存了，这里只是计数
        }
        fmt.Printf("Collected %d results.\n", totalResults)
        // 当output通道关闭时，表示所有结果都已收集
        close(done) // 通知调度器可以退出
    }()

    // 等待所有worker协程完成
    wg.Wait()
    close(output) // 所有worker都已退出，关闭output通道

    // 等待结果收集协程和调度器协程完成
    <-done
    fmt.Println("Crawl finished.")
}

func main() {
    urlList := []string{
        "http://example.com/page1",
        "http://example.com/page2",
        "http://example.com/fail1", // 模拟失败
        "http://example.com/page3",
        "http://example.com/page4",
        "http://example.com/fail2", // 模拟失败
        "http://example.com/page5",
    }
    crawl(urlList)
}

代码解析：

worker函数改动：
- 新增一个failed chan string参数。
- 当downloadURL返回非200状态码或错误时，不再向input通道发送URL，而是发送到failed通道：failed
- 引入*sync.WaitGroup来跟踪所有worker协程的完成状态，实现优雅停机。
crawl函数改动：
- 新增failed通道： failed := make(chan string)。
- 任务调度协程 (go func() {...})： 这是核心改动。
  - 它维护一个urlsToProcess切片，包含了所有待处理的URL（包括初始URL和从failed通道接收的URL）。
  - 使用一个select语句来同时监听两个事件：
    - input
    - url :=
  - 当urlsToProcess为空时，select语句会退化为只监听failed通道，避免了向空列表发送数据而导致的运行时错误。
  - 优雅退出机制：
    - done := make(chan struct{})：一个用于协调所有协程退出的通道。
    - 当output通道关闭后（表示所有worker都已处理完并退出了），结果收集协程会向done通道发送信号。
    - 调度器协程接收到done信号后，会关闭input通道，通知所有worker退出。
结果收集协程：
- 现在只负责从output通道接收数据。
- 当所有worker都完成并关闭output通道后，此协程会退出，并通过close(done)通知主调度流程。

这种设计确保了：

无死锁： worker协程永远不会阻塞在向input通道发送数据，因为它们只向failed通道发送，而failed通道由调度器协程负责消费。调度器协程的select语句保证了它不会因为尝试向空的input发送而阻塞，也不会因为没有failed任务而死等。
动态重试： 失败的URL可以被动态地重新加入待处理队列。
优雅停机： 通过sync.WaitGroup和done通道，可以确保所有worker协程、调度协程和结果收集协程都能在任务完成后安全退出。