代码之家 › 专栏 › 技术社区 › Terence Chow

如何处理可以在不阻塞的情况下增长的队列

goroutine channel go

Terence Chow · 技术社区 · 7 年前

我试图理解如果队列可以从处理函数本身增长,那么如何在Go中处理队列。见下面的代码。

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

sender 直到接收器接收到数据为止。在下面的代码中,每个进程都是处理新url的接收器。然而,很容易看出,如果一个进程向队列发送11个链接,它将阻塞,直到所有接收者处理完这些新链接。如果这些接收器每个都有一个链接,那么它们也会在将新的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 
}

2 回复 | 直到 7 年前

Tom Wright 7 年前

您可以使用的一个简单方法是将添加到通道的链接的代码移动到它自己的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 . 我绞尽脑汁想办法解决这个问题。可能是外部内存队列而不是通道?

Himanshu 7 年前

有两件事你可以做要么使用缓冲通道不阻止,即使没有人在另一端接收。这样就可以一次刷新通道中的值。

一种更有效的方法是检查通道中是否有任何可用的值,或者在发送所有值时由发送方关闭通道。

接收机可以通过指定接收表达式的第二个参数。

v, ok := <-ch

ok 是 false

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)
}

Playground example