用于取消机上例程的 golang 设计模式

Question

我是一个 golang 新手，正试图了解这个问题的正确设计模式。我当前的解决方案似乎非常冗长，而且我不确定更好的方法是什么。我正在尝试设计一个系统：执行 N 个协程一旦可用，就返回每个 goroutine 的结果如果一个 goroutine 返回一个特定的值，它应该杀死其他 goroutines 将取消。目标：我想启动多个 goroutine，但如果一个例程返回特定结果，我想取消这些例程。我试图了解我的代码是否超级“臭”，或者这是否是规定的做事方式。我仍然没有很好的感觉，所以任何帮助将不胜感激。这是我写的：package mainimport (    "context"    "fmt"    "time")func main() {    ctx := context.Background()    ctx, cancel := context.WithCancel(ctx)    fooCheck := make(chan bool)    barCheck := make(chan bool)    go foo(ctx, 3000, fooCheck)    go bar(ctx, 5000, barCheck)    for fooCheck != nil ||        barCheck != nil {        select {        case res, ok := <-fooCheck:            if !ok {                fooCheck = nil                continue            }            if res == false {                cancel()            }            fmt.Printf("result of foocheck: %t\n", res)        case res, ok := <-barCheck:            if !ok {                barCheck = nil                continue            }            fmt.Printf("result of barcheck: %t\n", res)        }    }    fmt.Printf("here we are at the end of the loop, ready to do some more processing...")}func foo(ctx context.Context, pretendWorkTime int, in chan<- bool) {    fmt.Printf("simulate doing foo work and pass ctx down to cancel down the calltree\n")    time.Sleep(time.Millisecond * time.Duration(pretendWorkTime))    select {    case <-ctx.Done():        fmt.Printf("\n\nWe cancelled this operation!\n\n")        break    default:        fmt.Printf("we have done some foo work!\n")        in <- false    }    close(in)}func bar(ctx context.Context, pretendWorkTime int, in chan<- bool) {    fmt.Printf("simulate doing bar work and pass ctx down to cancel down the calltree\n")    time.Sleep(time.Millisecond * time.Duration(pretendWorkTime))    select {    case <-ctx.Done():        fmt.Printf("\n\nWe cancelled the bar operation!\n\n")        break    default:        fmt.Printf("we have done some bar work!\n")        in <- true    }    close(in)}输出按预期工作，但恐怕我正在做一些决定，稍后会吹走我的脚。

Answer 1

我会使用单一渠道来传达结果，因此收集结果要容易得多，而且它会根据其性质自动“缩放”。如果您需要识别结果的来源，只需使用包含来源的包装器即可。是这样的：

type Result struct {

    ID     string

    Result bool

}

为了模拟“真实”工作，工作人员应该使用一个循环以迭代方式完成他们的工作，并且在每次迭代中他们应该检查取消信号。是这样的：

func foo(ctx context.Context, pretendWorkMs int, resch chan<- Result) {

    log.Printf("foo started...")

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

        time.Sleep(time.Millisecond)

        select {

        case <-ctx.Done():

            log.Printf("foo terminated.")

            return

        default:

        }

    }

    log.Printf("foo finished")

    resch <- Result{ID: "foo", Result: false}

}

在我们的示例中， thebar()是相同的，只是将所有foo单词替换为bar。

现在执行作业并提前终止其余作业（如果确实符合我们的预期），如下所示：

ctx, cancel := context.WithCancel(context.Background())

defer cancel()

resch := make(chan Result, 2)

log.Println("Kicking off workers...")

go foo(ctx, 3000, resch)

go bar(ctx, 5000, resch)

for i := 0; i < cap(resch); i++ {

    result := <-resch

    log.Printf("Result of %s: %v", result.ID, result.Result)

    if !result.Result {

        cancel()

        break

    }

}

log.Println("Done.")

运行此应用程序将输出（在Go Playground上尝试）：

2009/11/10 23:00:00 Kicking off workers...

2009/11/10 23:00:00 bar started...

2009/11/10 23:00:00 foo started...

2009/11/10 23:00:03 foo finished

2009/11/10 23:00:03 Result of foo: false

2009/11/10 23:00:03 Done.

有些事情要注意。如果我们由于意外结果提前终止，cancel()函数将被调用，我们从循环中跳出。可能其余的工作人员也同时完成他们的工作并发送他们的结果，这不会成为问题，因为我们使用了缓冲通道，所以他们的发送不会阻塞并且他们会正确结束。此外，如果他们没有同时完成，他们会检查ctx.Done()他们的循环，并且他们提前终止，所以 goroutines 被很好地清理了。

另请注意，上面代码的输出不打印bar terminated。这是因为main()函数在循环后立即终止，一旦main()函数结束，它不会等待其他非maingoroutines 完成。如果应用程序不会立即终止，我们也会看到该行被打印出来。time.Sleep()如果我们在末尾添加一个main()：

log.Println("Done.")

time.Sleep(3 * time.Millisecond)

输出将是（在Go Playground上尝试）：

2009/11/10 23:00:00 Kicking off workers...

2009/11/10 23:00:00 bar started...

2009/11/10 23:00:00 foo started...

2009/11/10 23:00:03 foo finished

2009/11/10 23:00:03 Result of foo: false

2009/11/10 23:00:03 Done.

2009/11/10 23:00:03 bar terminated.

现在，如果您必须等待所有工作人员“正常”或“提前”结束才能继续，您可以通过多种方式实现。

一种方法是使用sync.WaitGroup. 另一种方法是让每个工作人员发送一个Result无论他们如何结束，并且Result可以包含终止条件，例如normalor aborted。goroutinemain()可以继续接收循环，直到它n从 接收到值resch。如果选择此解决方案，您必须确保每个工作人员发送一个值（即使发生恐慌）以main()在这种情况下（例如使用 using defer）不阻塞。

Answer 2

我将针对您所说的内容分享最简单的模式。您可以将其扩展到更复杂的场景。

func doStuff() {

    // This can be a chan of anything.

    msgCh := make(chan string)

    // This is how you tell your go-routine(s) to stop, by closing this chan.

    quitCh := make(chan struct{})

    defer close(quitCh)

    // Start all go routines.

    for whileStart() {

        go func() {

            // Do w/e you need inside of your go-routine.

            // Write back the result.

            select {

            case msgCh <- "my message":

                // If we got here then the chan is open.

            case <-quitCh:

                // If we got here then the quit chan was closed.

            }

        }()

    }

    // Wait for all go routines.

    for whileWait() {

        // Block until a msg comes back.

        msg := <-msgCh

        // If you found what you want.

        if msg == stopMe {

            // It's safe to return because of the defer earlier.

            return

        }

    }

}