使用通道更快地關閉 goroutine

問題內容

我是 GO 新手，我有一個關於使用通道訊號停止 goroutine 的問題。

我有一個長期運行的 goroutine（超過 1000 個）和管理器來管理它：

func myThreadFunc(stop chan bool) {
    for {
        select {
        case <- stop:
            log.Debug("Stopping thread")
            return
        default:
            callClientTask() 
        }
    }
}

func callClientTask() {
    // This can take long time up to 30 seconds - this is external HTTP API call
    time.Sleep(5 * time.Second)
}


func manager() {
    var cancelChannelSlice []chan bool
    for i := 0; i < 1000; i++ {
        cancelChannel := make(chan bool)
        cancelChannelSlice = append(cancelChannelSlice, cancelChannel)

        go myThreadFunc(cancelChannel)
    }

    var stopTest = func() {
        for _, c := range cancelChannelSlice {
            c <- true
        }
    }

    timeout := time.After(time.Duration(300) * time.Second)
    for {
        select {
        case <-timeout:
            stopTest()
        default:
            time.Sleep(time.Second)
        }
    }
}

在這種情況下，每次我呼叫c <- true 管理員都會等待callClientTask() 完成，然後轉到下一個cancelChannel 我希望所有 goroutine 在 callClientTask() 的 1 次迭代中停止（不超過 30 秒）

我嘗試的唯一方法是像這樣投射新的 goroutine：

var stopTest = func() {
        for _, c := range cancelChannelSlice {
            go func(c chan bool) {
                c <- true
                close(c)
            }(c)
        }
    }

我這是正確的方法嗎？

正確答案

據我從您的問題中了解到，「您希望所有goroutine 在callClientTask() 的1 次迭代中停止（不超過30 秒）」且工作執行緒同時運行而不會出現同步問題。

我重新組織了與等待群組同時運行的程式碼。

範例程式碼：

#

package main

import (
    "log"
    "sync"
    "time"
)

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

    for {
        select {
        case <-stop:
            log.Println("Stopping thread")
            return
        default:
            callClientTask()
        }
    }
}

func callClientTask() {
    time.Sleep(2 * time.Second) // simulate a long-running task (for testing purposes)
}

func main() {
    var wg sync.WaitGroup
    stop := make(chan struct{})

    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go worker(stop, &wg)
    }

    time.Sleep(5 * time.Second) // allow workers to run for a while
    close(stop)                 // stop all workers, close channel
    wg.Wait()                   // wait for all workers
}

輸出：

2023/10/26 10:40:44 Stopping thread
2023/10/26 10:40:44 Stopping thread
....
2023/10/26 10:40:49 Stopping thread
2023/10/26 10:40:49 Stopping thread

編輯：

如果您想停止某些工作人員，則必須更新工作人員。以下程式碼包括具有“停止”和“停止”通道的工作人員以及啟動/停止功能。

範例程式碼：

#

package main

import (
    "log"
    "sync"
    "time"
)

type Worker struct {
    stop    chan struct{}
    stopped chan struct{}
}

func NewWorker() *Worker {
    return &Worker{
        stop:    make(chan struct{}),
        stopped: make(chan struct{}),
    }
}

func (w *Worker) Start(wg *sync.WaitGroup) {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for {
            select {
            case <-w.stop:
                log.Println("Stopping thread")
                close(w.stopped)
                return
            default:
                callClientTask()
            }
        }
    }()
}

func (w *Worker) Stop() {
    close(w.stop)
    <-w.stopped
}

func callClientTask() {
    time.Sleep(2 * time.Second) // simulate a long-running task (for testing purposes)
}

func main() {
    var wg sync.WaitGroup
    workers := make([]*Worker, 1000)

    for i := 0; i < 1000; i++ {
        workers[i] = NewWorker()
        workers[i].Start(&wg)
    }

    time.Sleep(5 * time.Second) // allow workers to run for a while 
    for i := 0; i < 100; i++ { // stop  first 100 workers
        workers[i].Stop()
    }  
    for i := 100; i < 1000; i++ { // wait other workers to finish
        workers[i].Stop()
    }
    wg.Wait()
}

輸出：

2023/10/26 12:51:26 Stopping thread
2023/10/26 12:51:28 Stopping thread
2023/10/26 12:51:30 Stopping thread
....

以上是使用通道更快地關閉 goroutine的詳細內容。更多資訊請關注PHP中文網其他相關文章！