package pipline // source部分 type ChanSource struct { out chan interface{} } // 定义一个新的channel源,包含状态收集，是否停止，和停止chan func NewSource(out chan interface{}) *ChanSource { return &ChanSource{ out: out, } } // Via streams data through the given flow func (cs *ChanSource) Via(flow Flow) Flow { go cs.DoStream(cs, flow) return flow } // DoStream 发送数据 outlet to inlet. func (cs *ChanSource) DoStream(outlet Outlet, inlet Inlet) { defer close(inlet.In()) for element := range outlet.Out() { inlet.In() <- element // fmt.Println("Source: ", element) } } // Out returns an output channel for sending data func (cs *ChanSource) Out() <-chan interface{} { return cs.out } 改造，达到优雅的关闭程序

可以在 ChanSource 结构体中新增一个字段，用于传递停止信号。同时，在 DoStream 方法中，使用 select 语句监听停止信号的接收，当接收到停止信号后，退出循环并关闭 inlet.In()。修改后的代码如下：

package pipeline

type ChanSource struct {
	out    chan interface{}
	stop   chan struct{}
}

func NewSource(out chan interface{}) *ChanSource {
	return &ChanSource{
		out:  out,
		stop: make(chan struct{}),
	}
}

func (cs *ChanSource) Via(flow Flow) Flow {
	go cs.DoStream(cs, flow)
	return flow
}

func (cs *ChanSource) DoStream(outlet Outlet, inlet Inlet) {
	defer close(inlet.In())
	for {
		select {
		case element, ok := <-outlet.Out():
			if !ok {
				return
			}
			inlet.In() <- element
		case <-cs.stop:
			return
		}
	}
}

func (cs *ChanSource) Out() <-chan interface{} {
	return cs.out
}

func (cs *ChanSource) Stop() {
	close(cs.stop)
}

然后，在程序退出前，调用 ChanSource 的 Stop 方法来通知关闭程序：

func main() {
	source := NewSource(nil)
	defer source.Stop()
	
	// ...
}