golang 扇入扇出以及通知退出机制

扇入和扇出通知退出机制golang实现生成器

最简单的带缓冲的生成器扇入技术实现增强型生成器退出通知机制实现自动退出的生成器一个融合了并发、缓冲、退出通知等多重特性的生成器

编程中经常遇到“扇入和扇出”两个概念，所谓的扇入是指将多路通道聚合到一条通道中处理，Go语言最简单的扇入就是使用select聚合多条通道服务;所谓的扇出是指将一条通道发散到多条通道中处理。在Go语言里面实现就是使用go关键字启动多个goroutine并发处理。

中国有句经典的哲学名句叫“分久必合，合久必分”，软件的设计和开发也遵循同样的哲学思想，扇入就是合，扇出就是分。当生产者的速度很慢时，需要使用扇入技术聚合多个生产者满足消费者，比如很耗时的加密/解密服务;当消费者的速度很慢时，需要使用扇出技术，比如Web服务器并发请求处理。扇入和扇出是Go并发编程中常用的技术。

读取已经关闭的通道不会引起阻塞，也不会导致panic，而是立即返回该通道存储类型的零值。关闭select监听的某个通道能使select立即感知这种通知，然后进行相应的处理，这就是所谓的退出通知机制(close channel to broadcast)。

其中context标准库就是利用这种机制处理更复杂的通知机制的。

在应用系统编程中，生成器一般用于生成全局事务号、订单号、序列号和随机数等等。

Go 对这种场景的支持非常简单，其中运用了扇入、扇出以及通知退出的思想。

package main

import (
	"fmt"
	"math/rand"
)

func GenerateIntA() chan int {
	ch := make(chan int, 10)
	// 启动一个goroutine用于生成随机数，函数返回一个通道用于获取随机数
	go func() {
		for {
			ch <- rand.Int()
		}
	}()
	return ch
}

func main() {
	ch := GenerateIntA()
	fmt.Println(<-ch)
	fmt.Println(<-ch)
}

//5577006791947779410
//8674665223082153551

package main

import (
	"fmt"
	"math/rand"
)

func GenerateIntA() chan int {
	ch := make(chan int, 10)
	go func() {
		for {
			ch <- rand.Int()
		}
	}()
	return ch
}

func GenerateIntB() chan int {
	ch := make(chan int, 10)
	go func() {
		for {
			ch <- rand.Int()
		}
	}()
	return ch
}

func GenerateInt() chan int {
	ch := make(chan int, 20)
	go func() {
		// 使用select的扇入技术(Fan in)增加生成的随机源
		for {
			select {
				case ch <- <-GenerateIntA():
				case ch <- <-GenerateIntB():
			}
		}
	}()
	return ch
}

func main() {
	ch := GenerateInt()
	for i := 0; i < 100; i++ {
		fmt.Println(<-ch)
	}
}

package main

import (
	"fmt"
	"math/rand"
)

func GenerateIntA(done chan struct{}) chan int {
	ch := make(chan int)
	go func() {
		Lable:
			for {
				// 通过select监听一个信号chan来确定是否停止生成
				select {
					case ch <- rand.Int():
					case <- done:
						break Lable
				}
			}
			close(ch)
	}()
	return ch
}

func main() {
	done := make(chan struct{})
	ch := GenerateIntA(done)

	fmt.Println(<-ch)
	fmt.Println(<-ch)

	// 不再需要生成器，通过close chan发送一个通知给生成器
	close(done)
	for v := range ch {
		fmt.Println(v)
	}
}

package main

import (
	"fmt"
	"math/rand"
)

// GenerateIntA done接收通知退出信号
func GenerateIntA(done chan struct{}) chan int {
	ch := make(chan int, 5)
	go func() {
		Lable:
			for {
				select {
					case ch <- rand.Int():
					case <- done:
						break Lable
				}
			}
			close(ch)
	}()
	return ch
}

// GenerateIntB done接收通知信号
func GenerateIntB(done chan struct{}) chan int {
	ch := make(chan int, 10)
	go func() {
	Lable:
		for {
			select {
			case ch <- rand.Int():
			case <- done:
				break Lable
			}
		}
		close(ch)
	}()
	return ch
}

// GenerateInt 通过select 执行扇入操作
func GenerateInt(done chan struct{}) chan int {
	ch := make(chan int)
	send := make(chan struct{})
	go func() {
		Lable:
			for {
				select {
					case ch <- <-GenerateIntA(send):
					case ch <- <-GenerateIntB(send):
					case <- done:
						send <- struct{}{}
						send <- struct{}{}
						break Lable
				}
			}
			close(ch)
	}()
	return ch
}

func main() {
	// 创建一个作为退出信号的chan
	done := make(chan struct{})

	// 启动生成器
	ch := GenerateInt(done)

	// 获取生成器资源
	for i := 0; i < 10; i++ {
		fmt.Println(<-ch)
	}
	// 通知生产者停止生产
	done <- struct{}{}
	fmt.Println("stop produce")
}

参考资料：

《Go核心编程》