当前位置： > > > > 如何处理可以增长而不会阻塞的队列

如何处理可以增长而不会阻塞的队列

来源：stackoverflow
2024-04-21 15:42:35
0浏览
收藏

本篇文章主要是结合我之前面试的各种经历和实战开发中遇到的问题解决经验整理的，希望这篇《如何处理可以增长而不会阻塞的队列》对你有很大帮助！欢迎收藏，分享给更多的需要的朋友学习~

问题内容

如果队列可以从处理函数本身增长，我试图了解如何在 go 中处理队列。请参阅下面的代码。

在此伪代码中，我想将正在创建的处理程序数量限制为 10 个。因此，我创建了 10 个处理队列的处理程序。然后我用一个 url 开始队列。

我的问题是，根据文档，通道的 sender 将阻塞，直到接收器收到数据。在下面的代码中，每个进程都是一个处理新 url 的接收者。然而，很容易看出，如果一个进程向队列发送 11 个链接，它将阻塞，直到所有接收者处理完这些新链接为止。如果这些接收器每个都有 1 个链接，那么它们在向队列发送新的 1 个链接时也会阻塞。由于每个人都被封锁，所以什么都没有完成。

我想知道 go 中的一般解决方案是什么，用于处理可以从进程本身增长的队列。请注意，我认为我可以通过锁定名为 queue 的数组来完成此操作，但我试图了解如何使用通道来完成此操作。

var queue = make(chan string)

func process(){
    for currentURL := range queue {
        links, _ := ... // some http call that gets links from a url
        for _, link := links {
            queue <- link
        }
    }
}

func main () {
   for i :=0; i < 10; i++ {
        go process()
   }

   queue <- "https://stackoverflow.com"
   ...
   // block until receive some quit message
   <-quit 
}

解决方案

您可以使用的一个简单方法是将添加通道链接的代码移动到它自己的 go 例程中。 这样，您的主要处理可以继续，而阻塞的通道写入则会阻塞单独的 go 例程。

func process(){
    for currenturl := range queue {
        links, _ := ... // some http call that gets links from a url
        for _, link := links {
            l := link // this is important! ...
            // the loop will re-set the value of link before the go routine is started

            go func(l) {
                queue <- link // we'll be blocked here...
                // but the "parent" routine can still iterate through the channel
                // which in turn un-blocks the write
            }(l)
        }
    }
}

使用信号量示例进行编辑以限制 go 例程：

func main () {
    maxworkers := 5000
    sem := semaphore.newweighted(int64(maxworkers))
    ctx := context.todo()
    for i :=0; i < 10; i++ {
        go process(ctx)
    }

    queue <- "https://stackoverflow.com"
    // block until receive some quit message
    <-quit 
}

func process(ctx context.context){
    for currenturl := range queue {
        links, _ := ... // some http call that gets links from a url
        for _, link := links {
            l := link // this is important! ...
            // the loop will re-set the value of link before the go routine is started

            // acquire a go routine...
            // if we are at the routine limit, this line will block until one becomes available
            sem.acquire(ctx, 1)
            go func(l) {
                defer sem.release(1)
                queue <- link // we'll be blocked here...
                // but the "parent" routine can still iterate through the channel
                // which in turn un-blocks the write
            }(l)
        }
    }
}

这个选项最终可能会导致死锁…假设所有 go 例程都已被声明，父循环可能会被锁定在 sem.acquire 上。这将导致子例程永远不会添加到通道，因此永远不会执行延迟的 sem.release。我正在努力想出一个好方法来处理这个问题。也许是外部内存队列而不是通道？

您可以做两件事，或者使用缓冲通道，即使另一端没有人接收，也不会阻塞。这样您就可以立即刷新通道内的值。

一种更有效的方法是检查通道中是否有任何可用值或通道是否已关闭，这应该在发送所有值时由发送者进行。

v, ok := <-ch

如果没有更多值要接收且通道已关闭，则 ok 为 false。使用 select as 检查通道内的值

package main

import (
    "fmt"
    "sync"
)

var queue = make(chan int)
var wg sync.WaitGroup

func process(){
        values := []int{1,2,5,3,9,7}
        for _, value := range values {
            queue <- value        
        }
}

func main () {
   for i :=0; i < 10; i++ {
        go process()
   }
   wg.Add(1)
   go func(){
      defer wg.Done()
      for j:=0;j<30;j++ {
          select {
             case <-queue:
        fmt.Println(<-queue)
          } 
      }
   }()
   wg.Wait()
   close(queue)
}

以上就是《如何处理可以增长而不会阻塞的队列》的详细内容，更多关于的资料请关注米云公众号！