如何有效防止通道死锁在递归工作池链上发生？

golang学习网今天将给大家带来《如何有效防止通道死锁在递归工作池链上发生？》，感兴趣的朋友请继续看下去吧！以下内容将会涉及到等等知识点，如果你是正在学习Golang或者已经是大佬级别了，都非常欢迎也希望大家都能给我建议评论哈~希望能帮助到大家！

假设您有一个基本的玩具系统，可以查找并处理目录中的所有文件（对于“进程”的某些定义）。其运作方式的基本图表如下所示：

如果这是一个现实世界的分布式系统，那么“箭头”实际上可以是无界队列，然后它就可以工作了。

在独立的 go 应用程序中，很容易将“箭头”建模为通道。然而，由于“通过需要列出子目录来生成更多工作”的自引用性质，很容易看出幼稚的实现会陷入僵局。例如（未经测试，请原谅编译错误）：

func ListDirWorker(dirs, files chan string) {
    for dir := range dirs {
        for _, path := range ListDir(dir) {
            if isDir(path) {
                    dirs <- path
                } else {
                    files <- path
                }
            }
        }
    }
}

如果我们想象我们只配置了一个列表工作器，那么只需要一个目录有两个子目录就可以基本上使这个事情陷入僵局。

我的大脑希望 golang 中有“无界通道”，但创建者不希望这样。建模这个东西的正确惯用方法是什么？我想有比实现线程安全队列并使用它而不是通道更简单的事情。 :)

正确答案

有一个非常相似的问题需要解决。需要：

• 有限数量的递归工作者 (bounded parallelism)
• content.Context 用于提前取消（强制执行超时限制等）
• 部分结果（一些 goroutine 遇到错误，而另一些则没有）
• 通过递归深度跟踪完成抓取（工作人员清理等）

下面我描述了问题以及我得出的解决方案的要点

问题：抓取不支持分页的 hr ldap 目录。服务器端限制还排除了超过 100k 记录的批量查询。需要小查询来解决这些限制。因此，从顶部（ceo）递归地导航树 - 列出员工（节点）并递归经理（分支）。

为了避免死锁 - 单个 workitem 通道不仅被工作线程用来读取（获取）工作，还被用来写入（委托）给其他空闲工作线程。这种方法可以使工人快速饱和。

注意：此处未包含但值得补充的是，使用通用 api 速率限制器来避免多个工作人员集体滥用/超过任何服务器端 api 速率限制。

要开始抓取，请创建工作线程并返回结果通道和错误通道。一些注意事项：

• c.in workitem 通道必须无缓冲才能使委派正常工作（稍后会详细介绍）
• c.rwg 跟踪所有工作线程的集体递归深度。当它达到零时，所有递归完成，爬行完成

func (c *crawler) crawl(ctx context.context, root branch, workers int) (<-chan result, <-chan error) {

    
    errc := make(chan error, 1)

    c.rwg = sync.waitgroup{}   // recursion depth waitgroup (to determine when all workers are done)
    c.rwg.add(1)               // add to waitgroups *before* starting workers
    c.in = make(chan workitem) // input channel: shared by all workers (to read from and also to write to when they need to delegate)
    c.out = make(chan result)  // output channel: where all workers write their results

    go func() {

        workererrc := c.createworkers(ctx, workers)

        c.in <- workitem{
            branch: root,   // initial place to start crawl
        }

        for err := range workererrc {
            if err != nil {
                // tally for partial results - or abort on first error (see werr)
            }
        }
        // summarize crawl success/failure via a single write to errc
        errc <- werr // nil, partial results, aborted early etc.
        close(errc)
    }

    return c.out, errc
}

创建有限数量的个体工人。返回的 error 通道收到每个单独工作线程的错误：

func (c *crawler) createworkers(ctx context.context, workers int) (<-chan error) {

    errc := make(chan error)

    var wg sync.waitgroup
    wg.add(workers)

    for i := 0; i < workers; i++ {
        i := i
        go func() {
            defer wg.done()

            var err error

            defer func() {
                errc <- err
            }()

            conn := dial("somewhere:8080") // worker prep goes here (open network connect etc.)

            for workitem := range c.in {
                err = c.recurse(ctx, i+1, conn, workitem)
                if err != nil {
                    return
                }
            }
        }()
    }

    go func() {
        c.rwg.wait() // wait for all recursion to finish ...
        close(c.in)  // ... so safe to close input channel ...
        wg.wait()    // ... wait for all workers to complete ...
        close(errc)    // .. finally signal to caller we're truly done
    }()

    return errc
}

recurse 逻辑：

• 对于任何可能阻塞的通道写入，请始终检查 ctx 是否取消，以便我们可以提前中止
• c.in 故意不缓冲，以确保委派正常工作（请参阅最后的注释）

func (c *Crawler) recurse(ctx context.Context, workID int, conn *net.Conn, wi workItem) error {

    defer c.rwg.Done() // decrement recursion count

    select {
    case <-ctx.Done():
        return ctx.Err()                      // canceled/timeout etc.

    case c.out <- Result{ /* Item: wi.. */}:  // write to results channel (manager or employee)
    }
    
    items, err := getItems(conn) // WORKER CODE (e.g. get manager employees etc.)
    if err != nil {
        return err
    }

    for _, i := range items {
        // leaf case
        if i.IsLeaf() {
            select {
            case <-ctx.Done():
                return ctx.Err()

            case c.out <- Result{ Item: i.Leaf }:
            }    
            continue
        }

        // branch case
        wi := workItem{
            branch: i.Branch,
        }

        c.rwg.Add(1) // about to recurse (or delegate-recursion)

        select {
        case c.in <- wi:
            // delegated to another worker!

        case <-ctx.Done(): // context canceled...
            c.rwg.Done() // ... so undo above `c.rwg.Add(1)`
            return ctx.Err()

        default:
            // no-one to delegated to (all busy) - so this worker will keep working
            err = c.recurse(ctx, workID, conn, wi)
            if err != nil {
                return err
            }
        }
    }

    return nil
}

授权是关键：

• 如果工作人员成功写入工作人员通道，则它知道工作已委托给另一个工作人员。
• 如果不能，那么工作人员就知道所有工作人员都在忙于工作（即不等待工作项） - 因此它必须自行递归

因此，人们既可以获得递归的好处，又可以利用固定大小的工作池。

终于介绍完啦！小伙伴们，这篇关于《如何有效防止通道死锁在递归工作池链上发生？》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！