Channel扇出模式

扇出模式

有扇入模式，就有扇出模式，扇出模式是和扇入模式相反的。扇出模式只有一个输入源 Channel，有多个目标 Channel，扇出比就是 1 比目标 Channel 数的值，经常用在设计模式中的观察者模式中（观察者设计模式定义了对象间的一种一对多的组合关系。这样一来，一个对象的状态发生变化时，所有依赖于它的对象都会得到通知并自动刷新）。在观察者模式中，数据变动后，多个观察者都会收到这个变更信号。下面是一个扇出模式的实现。从源 Channel 取出一个数据后，依次发送给目标 Channel。在发送给目标 Channel 的时候，可以同步发送，也可以异步发送：

// 扇出模式只有一个输入源 Channel，有多个目标 Channel，扇出比就是 1 比目标 Channel 数的值
func fanOut(ch <-chan interface{}, out []chan interface{}, async bool) {
    go func() {
        defer func() { //退出时关闭所有的输出chan
            for i := 0; i < len(out); i++ {
                close(out[i])
            }
        }()

        for v := range ch { // 从输入chan中读取数据
            v := v // 局部变量 v，以便在循环内部的闭包中使用
            for i := 0; i < len(out); i++ {
                i := i // 一个局部变量 i，以便在循环内部的闭包中使用
                if async {
                    go func() {
                        out[i] <- v // 放入到输出chan中,异步方式
                    }()
                } else {
                    out[i] <- v // 放入到输出chan中，同步方式
                }
            }
        }
    }()

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

reflectSelect 方式

// 实现扇出模式的反射方式,将数据从一个通道发送到多个目标通道
func fanOutReflect2(ch <-chan interface{}, out []chan interface{}) {
	// 输入通道输出通道同时在一个cases 中
	cases := make([]reflect.SelectCase, len(out)+1)

	// Create a case for the input channel
	// 接收通道，监听，有数据即可发送
	cases[0] = reflect.SelectCase{
		Dir:  reflect.SelectRecv, // case <-Chan: 从一个通道发出
		Chan: reflect.ValueOf(ch),
	}

	// Create cases for the output channels
	// 发送通道，监听，
	for i := 0; i < len(out); i++ {
		cases[i+1] = reflect.SelectCase{
			Dir:  reflect.SelectSend, // Chan <- Send： 发到多个目标通道
			Chan: reflect.ValueOf(out[i]),
		}
	}

	for {
		chosen, recv, _ := reflect.Select(cases)
		if chosen == 0 {
			// Data received from the input channel 从输入通道接收的数据
			v := recv.Interface()
			for i := 1; i < len(cases); i++ {
				// Send the data to all output channels 将数据发送到所有输出通道
				cases[i].Send = reflect.ValueOf(v)
			}
		} else {
			// Data sent to an output channel 发送到输出通道的数据
			// 把接收通道的数据置空
			cases[chosen].Send = reflect.Value{}
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37