func (cs *ChanSource) DoStream(outlet Outlet, inlet Inlet) { defer close(inlet.In()) for element := range outlet.Out() { select { case <-cs.cancel: return default: inlet.In() <- element // fmt.Println("Source: ", element) } // inlet.In() <- element // fmt.Println("Source: ", element) } } 提高性能

为了提高性能，可以通过两种方式来优化 DoStream() 函数：

1. 尽量减少 select 语句的使用。可以将 select 语句的判断放到 for 循环之前，从而减少 select 的使用次数。修改后的代码如下：

func (cs *ChanSource) DoStream(outlet Outlet, inlet Inlet) {
	defer close(inlet.In())
	select {
	case <-cs.cancel:
		return
	default:
		for element := range outlet.Out() {
			inlet.In() <- element
			// fmt.Println("Source: ", element)
			select {
			case <-cs.cancel:
				return
			default:
			}
		}
	}
}

2. 使用缓冲 channel。在新建 cancel chan 变量时，可以设置一个缓冲区，从而避免在写入 cancel 变量时发生阻塞。这样能够提高性能，减少资源浪费。修改后的代码如下：

// 定义一个新的channel源,包含状态收集，是否停止，和停止chan
func NewSource(out chan interface{}) *ChanSource {
	return &ChanSource{
		out:    out,
		cancel: make(chan bool, 1), // 设置缓冲区大小为1
	}
}

综上所述，上述两种方式都能够提升 DoStream() 函数的性能，具体选用哪种方式还需要根据具体的使用场景和数据流的大小来进行评估。