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

Question

我已经成功地做了一个同步解决方案，没有对调用的干扰。findMaxcompute

package main

import (

    "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，而不是灌浆计算函数调用的预期最大值。我尝试了不同的方法来做到这一点，并且大部分时间都陷入僵局。findMaxcomputectxmain

package main

import (

    "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

您的程序存在多个问题：您正在生成多个对共享变量进行操作的goroutine，即，并导致数据竞争，因为它们不受保护（例如，受Mutex保护）。maxnum这里由每个 worker goroutine 修改，但它应该是 worker 的本地，否则计算的数据可能会丢失（例如，一个 worker goroutine 计算了一个结果并将其存储在 num 中，但紧随其后，第二个 worker 计算并替换了 num 的值）。num&nbsp;num&nbsp;=&nbsp;compute&nbsp;//&nbsp;Should&nbsp;be&nbsp;"num&nbsp;:=&nbsp;compute"您不会等待每个 goroutine 完成计算，这可能会导致不正确的结果，因为即使上下文未取消，也不会考虑每个工作线程计算。使用 或 渠道来解决此问题。sync.WaitGroup下面是一个示例程序，可解决代码中的大多数问题：package mainimport (&nbsp; &nbsp; "context"&nbsp; &nbsp; "fmt"&nbsp; &nbsp; "math/rand"&nbsp; &nbsp; "sync"&nbsp; &nbsp; "time")type result struct {&nbsp; &nbsp; sync.RWMutex&nbsp; &nbsp; max uint64}func findMax(ctx context.Context, workers int) uint64 {&nbsp; &nbsp; var (&nbsp; &nbsp; &nbsp; &nbsp; res = result{}&nbsp; &nbsp; &nbsp; &nbsp; wg&nbsp; = sync.WaitGroup{}&nbsp; &nbsp; )&nbsp; &nbsp; for i := 0; i < workers; i++ {&nbsp; &nbsp; &nbsp; &nbsp; select {&nbsp; &nbsp; &nbsp; &nbsp; case <-ctx.Done():&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // RLock to read res.max&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; res.RLock()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ret := res.max&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; res.RUnlock()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return ret&nbsp; &nbsp; &nbsp; &nbsp; default:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; wg.Add(1)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; defer wg.Done()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; num := compute()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // Lock so that read from res.max and write&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // to res.max is safe. Else, data race could&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // occur.&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; res.Lock()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if num > res.max {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; res.max = num&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; res.Unlock()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }()&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }&nbsp; &nbsp; // Wait for all the goroutine to finish work i.e., all&nbsp; &nbsp; // workers are done computing and updating the max.&nbsp; &nbsp; wg.Wait()&nbsp; &nbsp; return res.max}func compute() uint64 {&nbsp; &nbsp; rnd := rand.Int63n(100)&nbsp; &nbsp; time.Sleep(time.Duration(rnd) * time.Millisecond)&nbsp; &nbsp; return rand.Uint64()}func main() {&nbsp; &nbsp; maxDuration := 2 * time.Second&nbsp; &nbsp; concurrency := 10&nbsp; &nbsp; ctx, cancel := context.WithTimeout(context.Background(), maxDuration)&nbsp; &nbsp; defer cancel()&nbsp; &nbsp; fmt.Println(findMax(ctx, concurrency))}正如@Brits在注释中指出的那样，当上下文被取消时，请确保停止那些工作线程goroutines以停止处理（如果可能的话），因为它不再需要了。