Home > Article > Backend Development > Close goroutines faster using channels
I am new to GO and I have a question about stopping a goroutine using a channel signal.
I have a long running goroutine (over 1000 of them) and a manager to manage it:
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) } } }
In this case, every time I call c <- true
the manager waits for callClientTask()
to complete before moving on to the next cancelChannel
I want all goroutines to stop in 1 iteration of callClientTask()
(no more than 30 seconds)
The only way I tried was to cast the new goroutine like this:
var stopTest = func() { for _, c := range cancelChannelSlice { go func(c chan bool) { c <- true close(c) }(c) } }
Am I doing this the right way?
As far as I understand from your question, "You want all goroutines to stop within 1 iteration of callClientTask() (no more than 30 seconds )" and the worker threads run simultaneously without synchronization issues.
I reorganized the code to run concurrently with the wait group.
Sample code:
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
edit:
If you want to stop some workers, you must update the workers. The following code includes a worker with "stop" and "stop" channels and a start/stop function.
Sample code:
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 ....
The above is the detailed content of Close goroutines faster using channels. For more information, please follow other related articles on the PHP Chinese website!