无法同时使用 goroutines 来查找最大值，直到上下文被取消

Question

我已经成功地做了一个同步解决方案，没有对调用的干扰。findMaxcomputepackage mainimport (    "context"    "fmt"    "math/rand"    "time")func findMax(ctx context.Context, concurrency int) uint64 {    var (        max uint64 = 0        num uint64 = 0    )    for i := 0; i < concurrency; i++ {        num = compute()        if num > max {            max = num        }    }    return max}func compute() uint64 {    // NOTE: This is a MOCK implementation of the blocking operation.        time.Sleep(time.Duration(rand.Int63n(100)) * time.Millisecond)    return rand.Uint64()}func main() {    maxDuration := 2 * time.Second    concurrency := 10    ctx, cancel := context.WithTimeout(context.Background(), maxDuration)    defer cancel()    max := findMax(ctx, concurrency)    fmt.Println(max)}https://play.golang.org/p/lYXRNTDtNCI当我尝试使用 goroutines 来重复调用函数时使用尽可能多的 goroutines，直到上下文被调用方函数取消。我每次都得到0，而不是灌浆计算函数调用的预期最大值。我尝试了不同的方法来做到这一点，并且大部分时间都陷入僵局。findMaxcomputectxmainpackage mainimport (    "context"    "fmt"    "math/rand"    "time")func findMax(ctx context.Context, concurrency int) uint64 {    var (        max uint64 = 0        num uint64 = 0    )    for i := 0; i < concurrency; i++ {        select {        case <- ctx.Done():            return max        default:            go func() {                num = compute()                if num > max {                    max = num                }            }()        }    }    return max}func compute() uint64 {    // NOTE: This is a MOCK implementation of the blocking operation.        time.Sleep(time.Duration(rand.Int63n(100)) * time.Millisecond)    return rand.Uint64()}func main() {    maxDuration := 2 * time.Second    concurrency := 10    ctx, cancel := context.WithTimeout(context.Background(), maxDuration)    defer cancel()    max := findMax(ctx, concurrency)    fmt.Println(max)}https://play.golang.org/p/3fFFq2xlXAE

Answer 1

您的程序存在多个问题：

1. 您正在生成多个对共享变量进行操作的goroutine，即，并导致数据竞争，因为它们不受保护（例如，受Mutex保护）。maxnum

2. 这里由每个 worker goroutine 修改，但它应该是 worker 的本地，否则计算的数据可能会丢失（例如，一个 worker goroutine 计算了一个结果并将其存储在 num 中，但紧随其后，第二个 worker 计算并替换了 num 的值）。num

 num = compute // Should be "num := compute"

1. 您不会等待每个 goroutine 完成计算，这可能会导致不正确的结果，因为即使上下文未取消，也不会考虑每个工作线程计算。使用 或 渠道来解决此问题。sync.WaitGroup

下面是一个示例程序，可解决代码中的大多数问题：

package main

import (

    "context"

    "fmt"

    "math/rand"

    "sync"

    "time"

)

type result struct {

    sync.RWMutex

    max uint64

}

func findMax(ctx context.Context, workers int) uint64 {

    var (

        res = result{}

        wg  = sync.WaitGroup{}

    )

    for i := 0; i < workers; i++ {

        select {

        case <-ctx.Done():

            // RLock to read res.max

            res.RLock()

            ret := res.max

            res.RUnlock()

            return ret

        default:

            wg.Add(1)

            go func() {

                defer wg.Done()

                num := compute()

                // Lock so that read from res.max and write

                // to res.max is safe. Else, data race could

                // occur.

                res.Lock()

                if num > res.max {

                    res.max = num

                }

                res.Unlock()

            }()

        }

    }

    // Wait for all the goroutine to finish work i.e., all

    // workers are done computing and updating the max.

    wg.Wait()

    return res.max

}

func compute() uint64 {

    rnd := rand.Int63n(100)

    time.Sleep(time.Duration(rnd) * time.Millisecond)

    return rand.Uint64()

}

func main() {

    maxDuration := 2 * time.Second

    concurrency := 10

    ctx, cancel := context.WithTimeout(context.Background(), maxDuration)

    defer cancel()

    fmt.Println(findMax(ctx, concurrency))

}

正如@Brits在注释中指出的那样，当上下文被取消时，请确保停止那些工作线程goroutines以停止处理（如果可能的话），因为它不再需要了。