使用通道停止长时间运行的方法

Question

我正在研究一个使用 Go 通道的示例，但遇到了一个问题。假设我们有一个运行大约 1 小时的方法。该方法执行各种操作，例如写入数据库、读取数据和发送电子邮件。我想让这个方法可以从外部中断，并且我正在尝试使用通道来实现这一点。

我遇到的问题是，由于没有收到来自 <-cancelled, the code inside the default case is being executed. However, the code inside the default case takes 5 seconds to execute, and I'm sending a cancelled signal from the main function at the 2nd second.

的信号

由于默认情况下的代码会继续执行，因此只有在默认块完成后才会捕获取消信号。

我们希望默认情况下的代码在收到取消信号时立即停止。

提前感谢您的帮助。祝你有美好的一天！

func worker(cancelled <-chan struct{}, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case <-cancelled:
            fmt.Println("Worker cancelled")
            return
        default:
            fmt.Println("Working...")
            time.Sleep(5 * time.Second)
            fmt.Println("ended")
        }
    }
}

func main() {
    cancelled := make(chan struct{})
    var wg sync.WaitGroup
    wg.Add(1)
    go worker(cancelled, &wg)

    go func() {
        time.Sleep(2 * time.Second)
        cancelled <- struct{}{}
    }()

    wg.Wait()
    fmt.Println("Program completed")
}

Answer 1

您应该使用

context.Context

而不是频道。事实上，这种操作是使用

context

包的常见模式之一。 Context 提供了使用相同上下文的 Done 通道来停止多个 Go 例程的能力。

select {
  case ctx.Done():
    return
  default:
    longRunningFunc(ctx)
}

如果

longRunningFunc

执行多个操作，则可以传递相同的上下文并让所有需要在收到信号时立即停止的函数监听。

但是如果长时间运行的操作是单个操作，那么就没有办法在中间停止它。在这种情况下，考虑这样的操作是否应该在信号上运行还是作为默认情况运行会很有用。

Answer 2

其实这种情况下，无论你使用context.Context还是channel，都需要自己处理可中断点。在每项工作开始之前检查状况，如下所示。这使您可以灵活地修改其中一项作业完成后想要执行的操作，但其他作业不会执行。

package main

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

func checkDone(done <-chan struct{}) bool {
    select {
    case <-done:
        fmt.Println("Worker cancelled")
        return true
    default:
        return false
    }
}

func worker(cancelled <-chan struct{}, wg *sync.WaitGroup) {
    defer wg.Done()

    for {
        if checkDone(cancelled) {
            return
        }

        fmt.Println("Working...")

        // Before important works start, check for cancellation
        if checkDone(cancelled) {
            return
        }

        time.Sleep(5 * time.Second)

        // this is interruptible point in the middle, check for cancellation
        if checkDone(cancelled) {
            // because previous job is done, 
            // but I receive the done signal, what should I do for this case?
            // write some code
            return
        }

        fmt.Println("ended")
    }
}

func main() {
    cancelled := make(chan struct{})
    var wg sync.WaitGroup
    wg.Add(1)
    go worker(cancelled, &wg)

    go func() {
        time.Sleep(2 * time.Second)
        cancelled <- struct{}{}
    }()

    wg.Wait()
    fmt.Println("Program completed")
}