errDiversion.jpg

只需简单指定upstreamChannel和errChannel即可开启一个eD。eD的工作逻辑如下：

1. 若errChan(以下简称为eC)已被关闭，则自己安全退出；
2. 若upstreamChan(以下简称为uC)已关闭，则将DATA直接丢弃；
3. 若upstreamChan处于开启状态，将DATA发送至uC。

判断uC的状态可以尝试关闭通道并捕获panic；或使用flag变量记录uC开闭状态即可(注意维护数据一致性)。

为什么要新创建一个eD goroutine而不是在子goroutine发送error前先作检查：

新建eD的过程应该在父goroutine完成的，并只需要传递给子goroutine一个用于传递err的channel(eC)即可。对子goroutine屏蔽细节。

再次使用作者项目作简单演示：

func errDiversion(eD *eDer) func(eC chan Errsocket) {
    //when upstream channel(uC) is closed(detected by closed flag),
    //following data will be received but discarded
    //when eC channel has closed, this goroutine will exit
    return func(eC chan Errsocket) {
        for {
            err, ok := <-eC
            if !ok { return }
            if !eD.closed {
                eD.eU <- err
            }
            eD.mu.Unlock()
        }
    }
}

func (s *server) Listen() { //Notifies the consumer when an error occurs ASYNCHRONOUSLY
    /*......*/
    //父goroutine创建eD，简单为子goroutine传递eC即可。
    s.eDerfunc = errDiversion(&s.eDer)  //closure
    eC := make(chan error)
    go s.eDerfunc(eC)
    go handleListen(s, eC)
    return
}

func handleListen(s *server, eC chan error) {
    /*......*/
    if err != nil {
        eC <- err    //s.errDoneTry即父goroutine传递的eC
    }
    /*......*/
}

建立函数闭包(closure)的原因：

对于一个eD而言，他的引用环境(uC，closed flag，mutex)是确定的。使用者只需传递eC即可使之正常工作。

数据一致性问题

最后简单提及维护数据一致性的问题。我们需要维护的有

1. flag与channel close的关系；
2. 确保eD能够及时执行(在uC关闭之前)【换言之，当eC存有error时，先等待eD处理error再关闭uC】。

在这里我们可以使用普通锁来实现：

type somestruct struct {
    //under protected data
    errDone    chan Errsocket
    errDoneTry chan Errsocket
    closed     bool

    mu         sync.Mutex

    /*...other data...*/
}

func xxx(){
    /*...*/
    //close upstream channel
    s.mu.Lock()
    s.closed = true
    close(s.errDone)
    s.mu.Unlock()
}

func subgoroutine(){
    /*...*/
    if err != nil{
        s.mu.Lock()    //Notice, lock before sending data to channel
        eC <- err
    }
}

errDiversion的代码也需要作部分调整：

func errDiversion(eD *eDer) func(eC chan Errsocket) {
    return func(eC chan Errsocket) {
        for {
            err, ok := <-eC
            if !ok {
                return
            }
            if !eD.closed {
                eD.eU <- err
            }
            eD.mu.Unlock()
        }
    }
}

这样我们就完成了并发流程控制以及数据的一致性。注意不要在eD中上锁，因为读取eC是一个阻塞过程，会引发死锁。正确的做法是向eC传递error之前上锁。

多eD嵌套的解决方案

即某上游eD(下简称为A)的eC是某下游eD（下简称为B）的uC。他们是共享同一个channel而非传递的关系。当B发送error至uC(a.eC)时，需要获得上游的锁并加锁。
为要实现该功能，将errDiversion代码改为

//......
if !closed{
    if eD.pmu != nil {
        eD.pmu.Lock() //send to upstream channel
    }
    eD.eU <- err
}
//......

即可。但在这么做之前务必三思，是否一定要使用嵌套的方法。事实上，你也可以将下游的uC与上游共享：即下文提到的1:m。

1:m或n:m的解决方案

即多个eD的共享同一个uC。同一个uC意味着同一把锁(Mutex)，同一个Flag标记uC状态。因此父goroutine应先将mutex和flag设置好，再将参数通过指针的方式传递给子goroutine。

等待错误反馈

上文提到的错误处理都是异步进行的。即不会等待错误处理，直接执行下面的流程。如果需要，可以阻塞代码流程等待错误反馈，即使错误处理同步进行。方法很简单，将mutex解锁的动作交与使用者即可。这样做的缺点也是显而易见的，使用者需要参与到库框架的运作中，增加了库运行的风险。同时，同步进行的错误处理也可能会降低运行效率。

总结

这套机制为处理goroutine异步通知error提供了一种有效解决方案，库的使用者仅需启用一个goroutine监听errchannel即可。更贴心的话，库还可以帮助使用者创建监听errchannel的框架，使使用者通过注入的方式填补框架即可。

不足之处在于：

1. 普遍情况下，开启一个子goroutine就需要另启一个eD作错误导流。过多的goroutine从性能而言并不是特别优秀；
2. 另外，他违反了通道关闭原则(一般原则下不允许接收方关闭通道和不能关闭一个有多个并发发送者的通道。 换而言之， 你只能在发送方的 goroutine 中关闭只有该发送方的通道)。关于如何优雅地关闭channel可参考这篇文章。

ps. 一套基于Socket的高并发Client-Server双向通信库chitchat-GitHub，目前还处于(缓慢悠闲的)开发状态。欢迎大家提供宝贵意见。