Tutup gorout dengan lebih pantas menggunakan saluran

Kandungan soalan

Saya baru untuk GO dan saya ada soalan tentang menghentikan goroutine menggunakan isyarat saluran.

Saya mempunyai goroutine yang telah lama berjalan (lebih 1000 daripadanya) dan seorang pengurus untuk menguruskannya:

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)
        }
    }
}

Dalam kes ini, setiap kali saya menelefon c <- true 管理器都会等待 callClientTask() 完成，然后转到下一个 cancelChannel Saya mahu semua goroutine berhenti dalam 1 lelaran callClientTask() (tidak lebih daripada 30 saat)

Satu-satunya cara yang saya cuba ialah membuang goroutine baharu seperti ini:

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

Adakah saya melakukan ini dengan cara yang betul?

Jawapan betul

Setakat yang saya faham daripada soalan anda, "Anda mahu semua goroutine berhenti dalam masa 1 lelaran callClientTask() (tidak lebih daripada 30 saat)" dan minta urutan pekerja dijalankan serentak tanpa soalan penyegerakan.

Saya menyusun semula kod untuk dijalankan serentak dengan kumpulan menunggu.

Kod sampel:

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
}

Output:

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

Editor:

Jika anda ingin menghentikan beberapa pekerja, anda perlu mengemas kini pekerja. Kod berikut termasuk pekerja dengan saluran "berhenti" dan "berhenti" serta fungsi mula/berhenti.

Kod sampel:

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()
}

Output:

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

Atas ialah kandungan terperinci Tutup gorout dengan lebih pantas menggunakan saluran. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!