一文教你Golang如何正确关闭通道

v, ok := <-ch
// 如果ok为true则代表通道已经关闭

func TestChanClosed(t *testing.T) {
	var ch = make(chan int)

	// send
	go func() {
		for {
			ch <- 1
		}
	}()

	// receive
	go func() {
		for {
			if v, ok := <-ch; ok {
				t.Log(v)
			} else {
				t.Log("通道关闭")
				return
			}
		}
	}()

	time.Sleep(1 * time.Second)
}

func TestChanClosed(t *testing.T) {
	var ch = make(chan int)

	// send
	go func() {
		for {
			ch <- 1
		}
	}()

	// receive
	go func() {
		for v := range ch {
			t.Log(v)
		}
		t.Log("通道关闭")
		return
	}()

	time.Sleep(1 * time.Second)
}

// 会引发channel panic的情况一:发送数据到已经关闭的channel
// panic: send on closed channel
func TestChannelPanic1(t *testing.T) {
	var ch = make(chan int)
	close(ch)
	time.Sleep(10 * time.Millisecond)
	go func() {
		ch <- 1
	}()
	t.Log(<-ch)
}

// 会引发channel panic的情况一的另外一种:发送数据时关闭channel
// panic: send on closed channel
func TestChannelPanic11(t *testing.T) {
	var ch = make(chan int)
	go func() {
		go func() {
			// 没有接收数据的地方,此处会一直阻塞
			ch <- 1
		}()
	}()

	time.Sleep(20 * time.Millisecond)
	close(ch)
}

// 会引发channel panic的情况二:重复关闭channel
// panic: close of closed channel
func TestChannelPanic2(t *testing.T) {
	var ch = make(chan int)
	close(ch)
	close(ch)
}

// 会引发channel panic的情况三:未初始化关闭
// panic: close of nil channel
func TestChannelPanic3(t *testing.T) {
	var ch chan int
	close(ch)
}

func TestSendClose(t *testing.T) {
	var (
		ch = make(chan int)
		wg = sync.WaitGroup{}
		// 10毫秒后通知发送端停止发送数据
		after = time.After(10 * time.Millisecond)
	)
	wg.Add(2)

	// send
	go func() {
		for {
			select {
			case <-after:
				close(ch)
				wg.Done()
				return
			default:
				ch <- 1
			}
		}
	}()

	// receive
	go func() {
		defer wg.Done()
		for v := range ch {
			t.Log(v)
		}
		return
	}()

	wg.Wait()
}

func TestManySendAndOneReceive(t *testing.T) {
	var (
		sender = 3
		wg     = sync.WaitGroup{}
		numCh  = make(chan int)
		stopCh = make(chan struct{})
		// 10毫秒后通知发送端停止发送数据
		after = time.After(10 * time.Millisecond)
	)
	wg.Add(1)

	// send
	for i := 0; i < sender; i++ {
		go func() {
			for {
				select {
				case <-stopCh:
					fmt.Println("收到退出信号")
					return
				case numCh <- 1:
					//fmt.Println("发送成功", value)
				}
			}
		}()
	}

	// receive
	go func() {
		for {
			select {
			case v := <-numCh:
				fmt.Println("接收到数据", v)
			case <-after:
				close(stopCh)
				wg.Done()
				return
			}
		}
	}()

	wg.Wait()
}

func TestManySendAndManyReceive(t *testing.T) {
	var (
		maxRandomNumber = 5000
		receiver        = 10
		sender          = 10
		wg              = sync.WaitGroup{}
		numCh           = make(chan int)
		stopCh          = make(chan struct{})
		toStop          = make(chan string, 1)
		stoppedBy       string
	)
	wg.Add(receiver)

	// moderator
	go func() {
		stoppedBy = <-toStop
		close(stopCh)
	}()

	// senders
	for i := 0; i < sender; i++ {
		go func(id string) {
			for {
				value := rand.Intn(maxRandomNumber)
				if value == 0 {
					select {
					case toStop <- "sender#" + id:
					default:
					}
					return
				}

				// 提前关闭goroutine
				select {
				case <-stopCh:
					return
				default:
				}

				select {
				case <-stopCh:
					return
				case numCh <- value:
				}
			}
		}(strconv.Itoa(i))
	}

	// receivers
	for i := 0; i < receiver; i++ {
		go func(id string) {
			defer wg.Done()
			for {
				// 提前关闭goroutine
				select {
				case <-stopCh:
					return
				default:
				}

				select {
				case <-stopCh:
					return
				case value := <-numCh:
					if value == maxRandomNumber-1 {
						select {
						case toStop <- "receiver#" + id:
						default:
						}
						return
					}

					t.Log(value)
				}
			}
		}(strconv.Itoa(i))
	}

	wg.Wait()
	t.Log("stopped by", stoppedBy)
}

type MyChannel struct {
	C    chan interface{}
	once sync.Once
}

func NewMyChannel() *MyChannel {
	return &MyChannel{C: make(chan interface{})}
}

func (mc *MyChannel) SafeClose() {
	mc.once.Do(func(){
		close(mc.C)
	})
}

type MyChannel struct {
	C      chan interface{}
	closed bool
	mutex  sync.Mutex
}

func NewMyChannel() *MyChannel {
	return &MyChannel{C: make(chan interface{})}
}

func (mc *MyChannel) SafeClose() {
	mc.mutex.Lock()
	if !mc.closed {
		close(mc.C)
		mc.closed = true
	}
	mc.mutex.Unlock()
}

func (mc *MyChannel) IsClosed() bool {
	mc.mutex.Lock()
	defer mc.mutex.Unlock()
	return mc.closed
}