从goroutine接收一定时间的值

Question

我有一个goroutine，可以生成无限数量的值（每个值比最后一个更合适），但是找到每个值需要越来越长的时间。我正试图找到一种方法来添加一个时间限制，比如10秒，之后我的函数会执行到目前为止收到的最佳值。

这是我目前的“解决方案”，使用频道和计时器：

// the goroutine which runs infinitely
// (or at least a very long time for high values of depth)
func runSearch(depth int, ch chan int) {
    for i := 1; i <= depth; i++ {
        fmt.Printf("Searching to depth %v\n", i)
        ch <- search(i)
    }
}

// consumes progressively better values until the channel is closed
func awaitBestResult(ch chan int) {
    var result int
    for result := range ch {
        best = result
    }

    // do something with best result here
}

// run both consumer and producer
func main() {
    timer := time.NewTimer(time.Millisecond * 2000)

    ch := make(chan int)

    go runSearch(1000, ch)
    go awaitBestResult(ch)

    <-timer.C
    close(ch)
}

这主要是有效的 - 在计时器结束和通道关闭后处理最佳结果。然而，然后我从panic: send on closed channel goroutine得到一个恐慌（runSearch），因为该频道已被主要功能关闭。

如何在计时器完成后停止第一个goroutine运行？非常感谢任何帮助。

Answer 1

您需要确保goroutine知道何时完成处理，以便它不会尝试写入已关闭的通道并发生混乱。

这听起来像context包的完美案例：

func runSearch(ctx context.Context, depth int, ch chan int) {
    for i := 1; i <= depth; i++ {
        select {
        case <- ctx.Done()
            // Context cancelled, return
            return
        default:
        }
        fmt.Printf("Searching to depth %v\n", i)
        ch <- search(i)
    }
}

然后在main()：

// run both consumer and producer
func main() {
    ctx := context.WithTimeout(context.Background, 2 * time.Second)

    ch := make(chan int)

    go runSearch(ctx, 1000, ch)
    go awaitBestResult(ch)

    close(ch)
}

Answer 2

你得到了恐慌，因为你的发送goroutine runSearch显然比计时器更长，并且它试图在已经被你的main goroutine关闭的通道上发送一个值。您需要设计一种方法来通知发送过程例程，一旦计时器失效并在关闭主要频道之前发送任何值。另一方面，如果你的搜索越早越好，你还需要与main沟通继续前进。您可以使用一个频道并进行同步，以便不存在竞争条件。最后，您需要知道消费者何时处理了所有数据，然后才能退出main。

这可能会有所帮助。

package main

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

var mu sync.Mutex //To protect the stopped variable which will decide if a value is to be sent on the signalling channel
var stopped bool

func search(i int) int {
    time.Sleep(1 * time.Millisecond)
    return (i + 1)
}

// (or at least a very long time for high values of depth)
func runSearch(depth int, ch chan int, stopSearch chan bool) {

    for i := 1; i <= depth; i++ {
        fmt.Printf("Searching to depth %v\n", i)
        n := search(i)
        select {
        case <-stopSearch:
            fmt.Println("Timer over! Searched till ", i)
            return
        default:
        }

        ch <- n
        fmt.Printf("Sent depth %v result for processing\n", i)
    }

    mu.Lock() //To avoid race condition with timer also being
    //completed at the same time as execution of this code
    if stopped == false {
        stopped = true
        stopSearch <- true
        fmt.Println("Search completed")
    }
    mu.Unlock()

}

// consumes progressively better values until the channel is closed
func awaitBestResult(ch chan int, doneProcessing chan bool) {
    var best int

    for result := range ch {
        best = result
    }
    fmt.Println("Best result ", best)
    // do something with best result here

    //and communicate to main when you are done processing the result
    doneProcessing <- true

}

func main() {
    doneProcessing := make(chan bool)
    stopSearch := make(chan bool)

    // timer := time.NewTimer(time.Millisecond * 2000)
    timer := time.NewTimer(time.Millisecond * 12)

    ch := make(chan int)

    go runSearch(1000, ch, stopSearch)
    go awaitBestResult(ch, doneProcessing)
    select {
    case <-timer.C:
        //If at the same time runsearch is also completed and trying to send a value !
        //So we hold a lock before sending value on the channel
        mu.Lock()
        if stopped == false {
            stopped = true
            stopSearch <- true
            fmt.Println("Timer expired")
        }
        mu.Unlock()

    case <-stopSearch:
        fmt.Println("runsearch goroutine completed")
    }
    close(ch)

    //Wait for your consumer to complete processing
    <-doneProcessing
    //Safe to exit now
}

在playground。更改timer的值以观察两种情况。