Golang模式：一次性终止多个goroutine

huangapple
go评论108阅读模式
英文:

Golang pattern to kill multiple goroutines at once

问题

我有两个goroutine，如下所示。我想要同步它们，以便当一个goroutine返回时，另一个goroutine也应该退出。在Go语言中，最好的方法是什么？

  1. func main() {
  3.   go func() {
  4.     ...
  5.     if err != nil {
  6.       return
  7.     }
  8.   }()
  10.   go func() {
  11.     ...
  12.     if err != nil {
  13.       return
  14.     }
  15.   }()
  18. }

我在这里模拟了这种情况https://play.golang.org/p/IqawStXt7rt，并尝试使用通道来信号化一个goroutine已完成。这看起来可能会导致对关闭的通道进行写入，从而导致恐慌。解决这个问题的最佳方法是什么？

英文:

I have two goroutines as shown in the snippet below. I want to synchronize them such that when one returns, the other one should also exit. What is best way in go to achieve this?

  1. func main() {
  3.   go func() {
  4.     ...
  5.     if err != nil {
  6.       return
  7.     }
  8.   }()
  10.   go func() {
  11.     ...
  12.     if err != nil {
  13.       return
  14.     }
  15.   }()
  18. }

I have simulated this scenario here https://play.golang.org/p/IqawStXt7rt and tried to solve it with a channel to signal a routine is done. This looks like there can be a write to closed channel resulting in a panic. What is the best way to solve this problem?

答案1

得分: 22

你可以使用上下文（context）在两个Go协程之间进行通信。

例如，

  1. package main
  3. import (
  4. 	"context"
  5. 	"sync"
  6. )
  8. func main() {
  10. 	ctx, cancel := context.WithCancel(context.Background())
  11. 	wg := sync.WaitGroup{}
  12. 	wg.Add(3)
  13. 	go func() {
  14. 		defer wg.Done()
  15. 		for {
  16. 			select {
  17. 			// 来自其他Go协程的消息完成
  18. 			case <-ctx.Done():
  19. 				// 结束
  20. 			}
  21. 		}
  22. 	}()
  24. 	go func() {
  25. 		defer wg.Done()
  26. 		for {
  27. 			select {
  28. 			// 来自其他Go协程的消息完成
  29. 			case <-ctx.Done():
  30. 				// 结束
  31. 			}
  32. 		}
  33. 	}()
  35. 	go func() {
  36. 		defer wg.Done()
  37. 		// 进行你的操作
  38. 		// 当这个Go协程结束时调用cancel
  39. 		cancel()
  40. 	}()
  41. 	wg.Wait()
  42. }
英文:

You can use context for communication between two go routines.
For example,

  1. package main
  3. import (
  4. 	&quot;context&quot;
  5. 	&quot;sync&quot;
  6. )
  8. func main() {
  10. 	ctx, cancel := context.WithCancel(context.Background())
  11. 	wg := sync.WaitGroup{}
  12. 	wg.Add(3)
  13. 	go func() {
  14. 		defer wg.Done()
  15. 		for {
  16. 			select {
  17. 			// msg from other goroutine finish
  18. 			case &lt;-ctx.Done():
  19. 				// end
  20. 			}
  21. 		}
  22. 	}()
  24. 	go func() {
  25. 		defer wg.Done()
  26. 		for {
  27. 			select {
  28. 			// msg from other goroutine finish
  29. 			case &lt;-ctx.Done():
  30. 				// end
  31. 			}
  32. 		}
  33. 	}()
  35. 	go func() {
  36. 		defer wg.Done()
  37. 		// your operation
  38. 		// call cancel when this goroutine ends
  39. 		cancel()
  40. 	}()
  41. 	wg.Wait()
  42. }

答案2

得分: 2

使用close在通道上发出完成信号。这允许多个goroutine通过在通道上接收来检查完成状态。

使用每个goroutine一个通道来表示goroutine的完成。

  1. done1 := make(chan struct{}) // 当goroutine 1返回时关闭
  2. done2 := make(chan struct{}) // 当goroutine 2返回时关闭
  4. go func() {
  5. 	defer close(done1)
  7.     timer1 := time.NewTicker(1 * time.Second)
  8.     defer timer1.Stop()
  10.     timer2 := time.NewTicker(2 * time.Second)
  11.     defer timer2.Stop()
  13. 	for {
  14. 		select {
  15. 		case <-done2:
  16.             // 另一个goroutine已返回。
  17. 			fmt.Println("done func 1")
  18. 			return
  19. 		case <-timer1.C:
  20. 			fmt.Println("timer1 func 1")
  21. 		case <-timer2.C:
  22. 			fmt.Println("timer2 func 1")
  23. 			return
  24. 		}
  26. 	}
  27. }()
  29. go func() {
  30. 	defer close(done2)
  31. 	for {
  32. 		select {
  33. 		case <-done1:
  34.             // 另一个goroutine已返回。
  35. 			fmt.Println("done func 2")
  36. 			return
  37. 		default:
  38. 			time.Sleep(3 * time.Second)
  39. 			fmt.Println("sleep done from func 2")
  40. 			return
  41. 		}
  43. 	}
  44. }()
  46. fmt.Println("等待goroutine完成")
  48. // 等待两个goroutine返回。这里等待的顺序无关紧要。
  49. <-done1
  50. <-done2
  52. fmt.Println("全部完成")

在playground上运行。

英文:

Use close on a channel to signal completion. This allows multiple goroutines to check for completion by receiving on the channel.

Use one channel per goroutine to signal completion of the goroutine.

  1. done1 := make(chan struct{}) // closed when goroutine 1 returns
  2. done2 := make(chan struct{}) // closed when goroutine 2 returns
  4. go func() {
  5. 	defer close(done1)
  7.     timer1 := time.NewTicker(1 * time.Second)
  8.     defer timer1.Stop()
  10.     timer2 := time.NewTicker(2 * time.Second)
  11.     defer timer2.Stop()
  13. 	for {
  14. 		select {
  15. 		case &lt;-done2:
  16.             // The other goroutine returned.
  17. 			fmt.Println(&quot;done func 1&quot;)
  18. 			return
  19. 		case &lt;-timer1.C:
  20. 			fmt.Println(&quot;timer1 func 1&quot;)
  21. 		case &lt;-timer2.C:
  22. 			fmt.Println(&quot;timer2 func 1&quot;)
  23. 			return
  24. 		}
  26. 	}
  27. }()
  29. go func() {
  30. 	defer close(done2)
  31. 	for {
  32. 		select {
  33. 		case &lt;-done1:
  34.             // The other goroutine returned.
  35. 			fmt.Println(&quot;done func 2&quot;)
  36. 			return
  37. 		default:
  38. 			time.Sleep(3 * time.Second)
  39. 			fmt.Println(&quot;sleep done from func 2&quot;)
  40. 			return
  41. 		}
  43. 	}
  44. }()
  46. fmt.Println(&quot;waiting for goroutines to complete&quot;)
  48. // Wait for both goroutines to return. The order that
  49. // we wait here does not matter. 
  50. &lt;-done1
  51. &lt;-done2
  53. fmt.Println(&quot;all done&quot;)

Run it on the playground.

答案3

得分: 2

首先，将等待goroutine和done通道分开。

使用sync.WaitGroup来协调goroutine。

  1. func main() {
  2. 	wait := &sync.WaitGroup{}
  3. 	N := 3
  5. 	wait.Add(N)
  6. 	for i := 1; i <= N; i++ {
  7. 		go goFunc(wait, i, true)
  8. 	}
  10. 	wait.Wait()
  11. 	fmt.Println(`Exiting main`)
  12. }

每个goroutine的代码如下：

  1. // 实际goroutine的代码
  2. func goFunc(wait *sync.WaitGroup, i int, closer bool) {
  3. 	defer wait.Done()
  4. 	defer fmt.Println(`Exiting `, i)
  6. 	T := time.Tick(time.Duration(100*i) * time.Millisecond)
  7. 	for {
  8. 		select {
  9. 		case <-T:
  10. 			fmt.Println(`Tick `, i)
  11. 			if closer {
  12. 				return
  13. 			}
  14. 		}
  15. 	}
  16. }

(https://play.golang.org/p/mDO4P56lzBU)

我们的main函数成功地在退出之前等待goroutine退出。每个goroutine都在关闭自身，我们希望有一种方法可以同时取消所有的goroutine。

我们将使用一个chan来实现这个，并利用通道接收的特性：

引用：关闭的通道上的接收操作总是可以立即进行，之前发送的任何值都已经被接收到后，会产生元素类型的零值。
(https://golang.org/ref/spec#Receive_operator)

我们修改我们的goroutine来检查CLOSE：

  1. func goFunc(wait *sync.WaitGroup, i int, closer bool, CLOSE chan struct{}) {
  2. 	defer wait.Done()
  3. 	defer fmt.Println(`Exiting `, i)
  5. 	T := time.Tick(time.Duration(100*i) * time.Millisecond)
  6. 	for {
  7. 		select {
  8. 		case <-CLOSE:
  9. 			return
  10. 		case <-T:
  11. 			fmt.Println(`Tick `, i)
  12. 			if closer {
  13. 				close(CLOSE)
  14. 			}
  15. 		}
  16. 	}
  17. }

然后我们改变我们的func main，使其传递CLOSE通道，并设置closer变量，以便只有最后一个goroutine会触发关闭：

  1. func main() {
  2. 	wait := &sync.WaitGroup{}
  3. 	N := 3
  4. 	CLOSE := make(chan struct{})
  6. 	// 启动goroutine
  7. 	wait.Add(N)
  8. 	for i := 1; i <= N; i++ {
  9. 		go goFunc(wait, i, i == N, CLOSE)
  10. 	}
  12. 	// 等待goroutine完成
  13. 	wait.Wait()
  14. 	fmt.Println(`Exiting main`)
  15. }

(https://play.golang.org/p/E91CtRAHDp2)

现在看起来一切都正常工作。

但实际上并不是这样的。并发是困难的。这段代码中潜伏着一个错误，只等着在生产环境中咬你一口。让我们找出它。

将我们的示例更改为每个goroutine都会关闭：

  1. func main() {
  2. 	wait := &sync.WaitGroup{}
  3. 	N := 3
  4. 	CLOSE := make(chan struct{})
  6. 	// 启动goroutine
  7. 	wait.Add(N)
  8. 	for i := 1; i <= N; i++ {
  9. 		go goFunc(wait, i, true /*** 每个goroutine都会关闭 ***/, CLOSE)
  10. 	}
  12. 	// 等待goroutine完成
  13. 	wait.Wait()
  14. 	fmt.Println(`Exiting main`)
  15. }

将goroutine更改为在关闭之前等待一段时间。我们希望两个goroutine在同一时间关闭：

  1. // 实际goroutine的代码
  2. func goFunc(wait *sync.WaitGroup, i int, closer bool, CLOSE chan struct{}) {
  3. 	defer wait.Done()
  4. 	defer fmt.Println(`Exiting `, i)
  6. 	T := time.Tick(time.Duration(100*i) * time.Millisecond)
  7. 	for {
  8. 		select {
  9. 		case <-CLOSE:
  10. 			return
  11. 		case <-T:
  12. 			fmt.Println(`Tick `, i)
  13. 			if closer {
  14. 				/*** 在关闭之前等待一段时间 ***/
  15. 				time.Sleep(time.Second)
  16. 				close(CLOSE)
  17. 			}
  18. 		}
  19. 	}
  20. }

(https://play.golang.org/p/YHnbDpnJCks)

我们遇到了崩溃：

  1. Tick  1
  2. Tick  2
  3. Tick  3
  4. Exiting  1
  5. Exiting  2
  6. panic: close of closed channel
  8. goroutine 7 [running]:
  9. main.goFunc(0x40e020, 0x2, 0x68601, 0x430080)
  10. 	/tmp/sandbox558886627/prog.go:24 +0x2e0
  11. created by main.main
  12. 	/tmp/sandbox558886627/prog.go:38 +0xc0
  14. Program exited: status 2.

虽然在关闭的通道上接收会立即返回，但是你不能关闭一个已经关闭的通道。

我们需要一些协调。我们可以使用sync.Mutex和一个bool来指示我们是否已经关闭了通道。让我们创建一个结构体来实现这个：

  1. type Close struct {
  2. 	C chan struct{}
  3. 	l sync.Mutex
  4. 	closed bool
  5. }
  7. func NewClose() *Close {
  8. 	return &Close {
  9. 		C: make(chan struct{}),
  10. 	}
  11. }
  13. func (c *Close) Close() {
  14. 	c.l.Lock()
  15. 	if (!c.closed) {
  16. 		c.closed=true
  17. 		close(c.C)
  18. 	}
  19. 	c.l.Unlock()
  20. }

重写我们的gofunc和main函数，使用我们的新Close结构体，然后我们就可以运行了：
https://play.golang.org/p/eH3djHu8EXW

并发的问题在于你总是需要考虑如果另一个“线程”在代码的任何其他地方会发生什么。

英文:

First separate the waiting on go-routines and the done channel.

Use a sync.WaitGroup to coordinate the goroutines.

  1. func main() {
  2. 	wait := &amp;sync.WaitGroup{}
  3. 	N := 3
  5. 	wait.Add(N)
  6. 	for i := 1; i &lt;= N; i++ {
  7. 		go goFunc(wait, i, true)
  8. 	}
  10. 	wait.Wait()
  11. 	fmt.Println(`Exiting main`)
  12. }

Each goroutine will look like this:

  1. // code for the actual goroutine
  2. func goFunc(wait *sync.WaitGroup, i int, closer bool) {
  3. 	defer wait.Done()
  4. 	defer fmt.Println(`Exiting `, i)
  6. 	T := time.Tick(time.Duration(100*i) * time.Millisecond)
  7. 	for {
  8. 		select {
  9. 		case &lt;-T:
  10. 			fmt.Println(`Tick `, i)
  11. 			if closer {
  12. 				return
  13. 			}
  14. 		}
  15. 	}
  16. }

(https://play.golang.org/p/mDO4P56lzBU)

Our main func is successfully waiting for the goroutines to exit before it exits. Each goroutine is closing itself, and we want a way to cancel all our goroutines at the same time.

We'll do this with a chan, and make use of this feature of receiving from channels:

QUOTE: A receive operation on a closed channel can always proceed immediately, yielding the element type's zero value after any previously sent values have been received.
(https://golang.org/ref/spec#Receive_operator)

We modify our goroutines to check for a CLOSE:

  1. func goFunc(wait *sync.WaitGroup, i int, closer bool, CLOSE chan struct{}) {
  2. 	defer wait.Done()
  3. 	defer fmt.Println(`Exiting `, i)
  5. 	T := time.Tick(time.Duration(100*i) * time.Millisecond)
  6. 	for {
  7. 		select {
  8. 		case &lt;-CLOSE:
  9. 			return
  10. 		case &lt;-T:
  11. 			fmt.Println(`Tick `, i)
  12. 			if closer {
  13. 				close(CLOSE)
  14. 			}
  15. 		}
  16. 	}
  17. }

and then we change our func main so that it passes the CLOSE channel through, and we'll set the closer variable so that only the last of our goroutines will trigger the close:

  1. func main() {
  2. 	wait := &amp;sync.WaitGroup{}
  3. 	N := 3
  4. 	CLOSE := make(chan struct{})
  6. 	// Launch the goroutines
  7. 	wait.Add(N)
  8. 	for i := 1; i &lt;= N; i++ {
  9. 		go goFunc(wait, i, i == N, CLOSE)
  10. 	}
  12. 	// Wait for the goroutines to finish
  13. 	wait.Wait()
  14. 	fmt.Println(`Exiting main`)
  15. }

(https://play.golang.org/p/E91CtRAHDp2)

Now it looks like everything is working.

But it isn't. Concurrency is hard. There's a bug lurking in this code, just waiting to bite you in production. Let's surface it.

Change our example so that every goroutine will close:

  1. func main() {
  2. 	wait := &amp;sync.WaitGroup{}
  3. 	N := 3
  4. 	CLOSE := make(chan struct{})
  6. 	// Launch the goroutines
  7. 	wait.Add(N)
  8. 	for i := 1; i &lt;= N; i++ {
  9. 		go goFunc(wait, i, true /*** EVERY GOROUTINE WILL CLOSE ***/, CLOSE)
  10. 	}
  12. 	// Wait for the goroutines to finish
  13. 	wait.Wait()
  14. 	fmt.Println(`Exiting main`)
  15. }

Change goroutine so that it takes a while before closing. We want two goroutines to be about to close at the same time:

  1. // code for the actual goroutine
  2. func goFunc(wait *sync.WaitGroup, i int, closer bool, CLOSE chan struct{}) {
  3. 	defer wait.Done()
  4. 	defer fmt.Println(`Exiting `, i)
  6. 	T := time.Tick(time.Duration(100*i) * time.Millisecond)
  7. 	for {
  8. 		select {
  9. 		case &lt;-CLOSE:
  10. 			return
  11. 		case &lt;-T:
  12. 			fmt.Println(`Tick `, i)
  13. 			if closer {
  14. 				/*** TAKE A WHILE BEFORE CLOSING ***/
  15. 				time.Sleep(time.Second)
  16. 				close(CLOSE)
  17. 			}
  18. 		}
  19. 	}
  20. }

(https://play.golang.org/p/YHnbDpnJCks)

We crash with:

  1. Tick  1
  2. Tick  2
  3. Tick  3
  4. Exiting  1
  5. Exiting  2
  6. panic: close of closed channel
  8. goroutine 7 [running]:
  9. main.goFunc(0x40e020, 0x2, 0x68601, 0x430080)
  10. 	/tmp/sandbox558886627/prog.go:24 +0x2e0
  11. created by main.main
  12. 	/tmp/sandbox558886627/prog.go:38 +0xc0
  14. Program exited: status 2.

While a receive on a closed channel returns immediately, you cannot close a closed channel.

We need a little coordination. We can do this with a sync.Mutex and a bool to indicate whether we've closed the channel or not. Let's create a struct to do this:

  1. type Close struct {
  2. 	C chan struct{}
  3. 	l sync.Mutex
  4. 	closed bool
  5. }
  7. func NewClose() *Close {
  8. 	return &amp;Close {
  9. 		C: make(chan struct{}),
  10. 	}
  11. }
  13. func (c *Close) Close() {
  14. 	c.l.Lock()
  15. 	if (!c.closed) {
  16. 		c.closed=true
  17. 		close(c.C)
  18. 	}
  19. 	c.l.Unlock()
  20. }

Rewrite our gofunc and our main to use our new Close struct, and we're good to go:
https://play.golang.org/p/eH3djHu8EXW

The problem with concurrency is that you always need to be wondering what would happen if another 'thread' was anywhere else in the code.

答案4

得分: 1

你的问题是你希望在"完成"通道上发送一次，但却被多个监听器接收。你还需要考虑发送到done通道的是由goroutine接收还是由main函数接收。

我建议你将等待goroutine和done通道分开。

  1. import `sync`
  3. // 这段代码将在两个函数完成之前等待，然后结束
  4. func main {
  5.    var wait sync.WaitGroup
  6.    wait.Add(2)
  7.    go func() {
  8.      defer wait.Done()
  9.    }()
  10.    go g() {
  11.      defer wait.Done()
  12.    }()
  13.    wait.Wait()
  14. }

现在，如何管理Done。解决方案是使用sync.Cond，并让每个goroutine运行自己的goroutine来等待Cond。以下是一个示例：

  1. package main
  3. import (
  4. 	`fmt`
  5. 	`sync`
  6. 	`time`
  7. )
  9. // WaitForIt 封装了一个Cond和一个Mutex，提供了一个更简单的API：
  10. // .WAIT() chan struct{} 返回一个通道，当WaitForIt完成时会发出信号。
  11. // .Done() 表示WaitForIt完成。
  12. type WaitForIt struct {
  13. 	L *sync.Mutex
  14. 	Cond *sync.Cond
  15. }
  17. func NewWaitForIt() *WaitForIt {
  18. 	l := &amp;sync.Mutex{}
  19. 	c := sync.NewCond(l)
  20. 	return &amp;WaitForIt{ l, c }
  21. }
  23. // WAIT 返回一个通道，当Cond触发时会发出信号。
  24. func (w *WaitForIt) WAIT() chan struct{} {
  25. 	D := make(chan struct{})
  26. 	go func() {
  27. 		w.L.Lock()
  28. 		defer w.L.Unlock()
  29. 		w.Cond.Wait()
  30. 		D &lt;- struct{}{}
  31. 		close(D)
  32. 	}()
  33. 	return D
  34. }
  36. // Done 表示Cond应该被触发。
  37. func (w *WaitForIt) Done() {
  38. 	w.Cond.Broadcast()
  39. }
  41. // doneFunc 使用一个通道来启动函数f，当函数应该停止时，通道会发出信号。
  42. // 它还处理WaitGroup的同步。
  43. func doneFunc(wait *sync.WaitGroup, waitForIt *WaitForIt, f func(DONE chan struct{})) {
  44. 	defer wait.Done()
  45. 	f(waitForIt.WAIT())
  46. }
  48. func main() {
  49. 	// wait 将在main()函数级别协调所有的goroutine
  50.     // waitForIt 将协调所有的goroutine
  51. 	wait := &amp;sync.WaitGroup{}
  52. 	// waitForIt 用于指示goroutine何时关闭
  53. 	waitForIt := NewWaitForIt()
  55.     // goFunc 生成每个goroutine。只有3秒的goroutine会关闭所有的goroutine
  56. 	goFunc := func(seconds int) func(chan struct{}) {
  57. 		return func(DONE chan struct{}) {
  58.             // 这是每个goroutine的实际代码
  59.             // 它创建一个计时器，持续一定的秒数，
  60.             // 在计时器结束后打印秒数，
  61.             // 或者在DONE被触发时退出
  62. 			timer := time.NewTicker(time.Duration(seconds) * time.Second)
  63. 			defer timer.Stop()
  64. 			for {
  65. 				select {
  66. 				case &lt;- DONE:
  67. 					return
  68. 				case &lt;- timer.C:
  69. 					if (3==seconds) {
  70. 						waitForIt.Done()
  71.                         // 不要在这里关闭 - 我们将在DONE被触发时关闭
  72. 					}
  73. 				}
  74. 			}
  75. 		}
  76. 	}
  77. 	// 启动3个goroutine，每个都在等待关闭信号
  78. 	for i:=1; i&lt;=3; i++ {
  79. 		wait.Add(1)
  80. 		go doneFunc(wait, waitForIt, goFunc(i))
  81. 	}
  82. 	// 等待所有的goroutine完成，然后我们完成了
  83. 	wait.Wait()
  84. }

这是使用WaitForIt实现的示例：https://play.golang.org/p/llphW73G1xE
请注意，我不得不在WaitForIt.Done中删除Lock()调用。尽管文档说你可以持有锁，但它会阻塞第二个goroutine的完成。

英文:

Your problem is that you want a single send on the DONE channel to be received by multiple listeners. You also need to consider whether a send on the done channel is received by your goroutines, or by your main func.

I suggest you rather separate the waiting on go-routines and the done channel.

  1. import `sync`
  3. // This code will wait for the two functions to complete before ending
  4. func main {
  5.    var wait sync.WaitGroup
  6.    wait.Add(2)
  7.    go func() {
  8.      defer wait.Done()
  9.    }()
  10.    go g() {
  11.      defer wait.Done()
  12.    }()
  13.    wait.Wait()
  14. }

Now, how to manage the Done. Well, the solution is to use a sync.Cond and have each goroutine run its own goroutine to wait on the Cond. Here's an example:

  1. package main
  3. import (
  4. 	`fmt`
  5. 	`sync`
  6. 	`time`
  7. )
  9. // WaitForIt wraps a Cond and a Mutex for a simpler API:
  10. // .WAIT() chan struct{} will return a channel that will be
  11. //   signalled when the WaitForIt is done.
  12. // .Done() will indicate that the WaitForIt is done.
  13. type WaitForIt struct {
  14. 	L *sync.Mutex
  15. 	Cond *sync.Cond
  16. }
  18. func NewWaitForIt() *WaitForIt {
  19. 	l := &amp;sync.Mutex{}
  20. 	c := sync.NewCond(l)
  21. 	return &amp;WaitForIt{ l, c }
  22. }
  24. // WAIT returns a chan that will be signalled when
  25. // the Cond is triggered.
  26. func (w *WaitForIt) WAIT() chan struct{} {
  27. 	D := make(chan struct{})
  28. 	go func() {
  29. 		w.L.Lock()
  30. 		defer w.L.Unlock()
  31. 		w.Cond.Wait()
  32. 		D &lt;- struct{}{}
  33. 		close(D)
  34. 	}()
  35. 	return D
  36. }
  38. // Done indicates that the Cond should be triggered.
  39. func (w *WaitForIt) Done() {
  40. 	w.Cond.Broadcast()
  41. }
  43. // doneFunc launches the func f with a chan that will be signalled when the
  44. // func should stop. It also handles WaitGroup synchronization
  45. func doneFunc(wait *sync.WaitGroup, waitForIt *WaitForIt, f func(DONE chan struct{})) {
  46. 	defer wait.Done()
  47. 	f(waitForIt.WAIT())
  48. }
  50. func main() {
  51. 	// wait will coordinate all the goroutines at the level of main()
  52.     // between themselves the waitForIt will do the coordination
  53. 	wait := &amp;sync.WaitGroup{}
  54. 	// waitForIt indicates to the goroutines when they should shut
  55. 	waitForIt := NewWaitForIt()
  57.     // goFunc generates each goroutine. Only the 3-second goroutine will 
  58.     // shutdown all goroutines
  59. 	goFunc := func(seconds int) func(chan struct{}) {
  60. 		return func(DONE chan struct{}) {
  61.             // this is the actual code of each goroutine
  62.             // it makes a ticker for a number of seconds,
  63.             // and prints the seconds after the ticker elapses,
  64.             // or exits if DONE is triggered
  65. 			timer := time.NewTicker(time.Duration(seconds) * time.Second)
  66. 			defer timer.Stop()
  67. 			for {
  68. 				select {
  69. 				case &lt;- DONE:
  70. 					return
  71. 				case &lt;- timer.C:
  72. 					if (3==seconds) {
  73. 						waitForIt.Done()
  74.                         // Don&#39;t shutdown here - we&#39;ll shutdown
  75.                         // when our DONE is signalled
  76. 					}
  77. 				}
  78. 			}
  79. 		}
  80. 	}
  81. 	// launch 3 goroutines, each waiting on a shutdown signal
  82. 	for i:=1; i&lt;=3; i++ {
  83. 		wait.Add(1)
  84. 		go doneFunc(wait, waitForIt, goFunc(i))
  85. 	}
  86. 	// wait for all the goroutines to complete, and we&#39;re done
  87. 	wait.Wait()
  88. }

Here's your example implemented using WaitForIt: https://play.golang.org/p/llphW73G1xE
Note that I had to remove the Lock() call in WaitForIt.Done. Although the documentation says you're allowed to hold the lock, it was blocking your 2nd goroutine from completing.

答案5

得分: 0

  1. package main
  3. import (
  4. 	"fmt"
  5. 	"sync"
  6. 	"time"
  7. )
  9. func func1(done chan struct{}, wg *sync.WaitGroup) {
  10. 	defer wg.Done()
  11. 	timer1 := time.NewTicker(1 * time.Second)
  12. 	timer2 := time.NewTicker(2 * time.Second)
  13. 	for {
  14. 		select {
  15. 		case <-timer1.C:
  16. 			fmt.Println("定时器1函数1")
  17. 		case <-timer2.C:
  18. 			// 请求GC清理定时器timer1和timer2
  19. 			// 因为goroutine即将返回
  20. 			timer1.Stop()
  21. 			timer2.Stop()
  23. 			fmt.Println("定时器2函数1")
  25. 			done <- struct{}{} // 通知另一个goroutine终止
  27. 			fmt.Println("从函数1发送done")
  28. 			return
  29. 		case <-done:
  30. 			// 请求GC清理定时器timer1和timer2
  31. 			// 因为goroutine即将返回
  32. 			timer1.Stop()
  33. 			timer2.Stop()
  35. 			fmt.Println("函数1完成")
  36. 			return
  38. 		}
  40. 	}
  41. }
  43. func func2(done chan struct{}, wg *sync.WaitGroup) {
  44. 	defer wg.Done()
  45. 	timer3 := time.NewTicker(3 * time.Second)
  46. 	for {
  47. 		select {
  48. 		case <-timer3.C:
  49. 			// 请求GC清理定时器timer3
  50. 			// 因为goroutine即将返回
  51. 			timer3.Stop()
  53. 			fmt.Println("定时器3函数2")
  55. 			done <- struct{}{} // 通知另一个goroutine终止
  57. 			fmt.Println("从函数2发送done")
  58. 			return
  59. 		case <-done:
  60. 			// 请求GC清理定时器timer3
  61. 			// 因为goroutine即将返回
  62. 			timer3.Stop()
  63. 			fmt.Println("函数2完成")
  64. 			return
  65. 		}
  67. 	}
  68. }
  70. func main() {
  71. 	// 用于goroutine之间的信号通信的通道
  72. 	done := make(chan struct{})
  74. 	// WaitGroup
  75. 	wg := sync.WaitGroup{}
  77. 	wg.Add(2)
  79. 	// 启动func1的goroutine
  80. 	go func1(done, &wg)
  81. 	// 启动func2的goroutine
  82. 	go func2(done, &wg)
  84. 	fmt.Println("开始休眠")
  86. 	// 等待goroutine完成
  87. 	wg.Wait()
  89. 	// 等待15秒钟
  90. 	// 如果不需要，请删除下面的代码
  91. 	time.Sleep(15 * time.Second)
  92. 	fmt.Println("等待了15秒钟")
  94. }
英文: 
  1. package main
  3. import (
  4. 	&quot;fmt&quot;
  5. 	&quot;sync&quot;
  6. 	&quot;time&quot;
  7. )
  9. func func1(done chan struct{}, wg *sync.WaitGroup) {
  10. 	defer wg.Done()
  11. 	timer1 := time.NewTicker(1 * time.Second)
  12. 	timer2 := time.NewTicker(2 * time.Second)
  13. 	for {
  14. 		select {
  15. 		case &lt;-timer1.C:
  16. 			fmt.Println(&quot;timer1 func 1&quot;)
  17. 		case &lt;-timer2.C:
  18. 			// Ask GC to sweep the tickers timer1, timer2
  19. 			// as goroutine should return
  20. 			timer1.Stop()
  21. 			timer2.Stop()
  23. 			fmt.Println(&quot;timer2 func 1&quot;)
  25. 			done &lt;- struct{}{} // Signal the other goroutine to terminate
  27. 			fmt.Println(&quot;sent done from func 1&quot;)
  28. 			return
  29. 		case &lt;-done:
  30. 			// Ask GC to sweep the tickers timer1, timer2
  31. 			// as goroutine should return
  32. 			timer1.Stop()
  33. 			timer2.Stop()
  35. 			fmt.Println(&quot;done func 1&quot;)
  36. 			return
  38. 		}
  40. 	}
  41. }
  43. func func2(done chan struct{}, wg *sync.WaitGroup) {
  44. 	defer wg.Done()
  45. 	timer3 := time.NewTicker(3 * time.Second)
  46. 	for {
  47. 		select {
  48. 		case &lt;-timer3.C:
  49. 			// Ask GC to sweep the tickers timer3
  50. 			// as goroutine should return
  51. 			timer3.Stop()
  53. 			fmt.Println(&quot;timer3 func 2&quot;)
  55. 			done &lt;- struct{}{} // Signal the other goroutine to terminate
  57. 			fmt.Println(&quot;sent done from func 2&quot;)
  58. 			return
  59. 		case &lt;-done:
  60. 			// Ask GC to sweep the tickers timer3
  61. 			// as goroutine should return
  62. 			timer3.Stop()
  63. 			fmt.Println(&quot;done func 2&quot;)
  64. 			return
  65. 		}
  67. 	}
  68. }
  70. func main() {
  71. 	// Chan used for signalling between goroutines
  72. 	done := make(chan struct{})
  74. 	// WaitGroup
  75. 	wg := sync.WaitGroup{}
  77. 	wg.Add(2)
  79. 	// Spawn the goroutine for func1
  80. 	go func1(done, &amp;wg)
  81. 	// Spawn the goroutine for func2
  82. 	go func2(done, &amp;wg)
  84. 	fmt.Println(&quot;starting sleep&quot;)
  86. 	// Wait for the goroutines
  87. 	wg.Wait()
  89. 	// Wait for 15 seconds
  90. 	// If not required, please remove
  91. 	// the lines below
  92. 	time.Sleep(15 * time.Second)
  93. 	fmt.Println(&quot;waited 15 seconds&quot;)
  95. }

答案6

得分: 0

你可以使用通道关闭模式在多个Go协程中等待。

  1. package main
  3. import (
  4. 	"os"
  5. 	"os/signal"
  6. 	"sync"
  7. 	"syscall"
  8. )
  10. type RunGroup struct {
  11. 	sync.WaitGroup
  12. }
  14. // Run处理等待组状态
  15. func (runGroup *RunGroup) Run(f func()) {
  16. 	runGroup.Add(1)
  17. 	go func() {
  18. 		f()
  19. 		runGroup.Done()
  20. 	}()
  21. }
  23. func doStuff(done <-chan any, id int) {
  24. 	println("Doing something", id)
  25. 	<-done
  26. 	println("DONE", id)
  27. }
  29. func main() {
  30. 	// Done通道
  31. 	done := make(chan any)
  32. 	// 设置关闭监听器
  33. 	sigChan := make(chan os.Signal, 1)
  34. 	signal.Notify(sigChan, syscall.SIGTERM)
  35. 	signal.Notify(sigChan, syscall.SIGINT)
  36. 	go func() {
  37. 		rawSig := <-sigChan
  38. 		sig := rawSig.String()
  39. 		println("Caught signal, shutting down.", sig)
  40. 		close(done)
  41. 	}()
  43. 	runGroup := RunGroup{}
  44. 	// 做一些事情
  45. 	runGroup.Run(func() {
  46. 		doStuff(done, 1)
  47. 	})
  48. 	runGroup.Run(func() {
  49. 		doStuff(done, 2)
  50. 	})
  51. 	runGroup.Run(func() {
  52. 		doStuff(done, 3)
  53. 	})
  55. 	// 主线程等待中断
  56. 	runGroup.Wait()
  57. }

输出

  1. go run ./main.go
  2. Doing something 3
  3. Doing something 2
  4. Doing something 1
  5. ^CCaught signal, shutting down. interrupt
  6. DONE 3
  7. DONE 1
  8. DONE 2
英文:

You could use channel closing pattern to wait within multiple go routines.

  1. package main
  3. import (
  4. 	&quot;os&quot;
  5. 	&quot;os/signal&quot;
  6. 	&quot;sync&quot;
  7. 	&quot;syscall&quot;
  8. )
  11. type RunGroup struct {
  12.   sync.WaitGroup
  13. }
  15. // Run handles wait group state
  16. func (runGroup *RunGroup) Run(f func()) {
  17.   runGroup.Add(1)
  18.   go func() {
  19.     f()
  20.     runGroup.Done()
  21.   }()
  22. }
  24. func doStuff(done &lt;-chan any, id int) {
  25.   println(&quot;Doing something&quot;, id)
  26.   &lt;-done
  27.   println(&quot;DONE&quot;, id)
  28. }
  30. func main() {
  31.   // Done channel
  32.   done := make(chan any)
  33.   // Setup Shutdown listeners
  34.   sigChan := make(chan os.Signal, 1)
  35. 	signal.Notify(sigChan, syscall.SIGTERM)
  36. 	signal.Notify(sigChan, syscall.SIGINT)
  37.   go func() {
  38.     rawSig := &lt;-sigChan
  39.     sig := rawSig.String()
  40.     println(&quot;Caught signal, shutting down.&quot;, sig)
  41.     close(done)
  42.   }()
  44.   runGroup := RunGroup{}
  45.   // Do some stuff
  46.   runGroup.Run(func () {
  47.     doStuff(done, 1)
  48.   })
  49.   runGroup.Run(func () {
  50.     doStuff(done, 2)
  51.   })
  52.   runGroup.Run(func () {
  53.     doStuff(done, 3)
  54.   })
  56.   // Wait mainthread until interrupt
  57.   runGroup.Wait()
  58. }

Output

  1. go run ./main.go
  2. Doing something 3
  3. Doing something 2
  4. Doing something 1
  5. ^CCaught signal, shutting down. interrupt
  6. DONE 3
  7. DONE 1
  8. DONE 2

huangapple
  • 本文由 发表于 2020年4月3日 15:57:27
  • 转载请务必保留本文链接：https://go.coder-hub.com/61007385.html
  • channel
  • go
  • goroutine
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定