设计Go API的管道使用原则

管道是并发安全的队列，用于在Go的轻量级线程(Go协程)之间安全地传递消息。总的来讲，这些原语是Go语言中最为称道的特色功能之一。这种消息传递范式使得开发者可以以易于理解的语义和控制流来协调管理多线程并发任务，而这胜过使用回调函数或者共享内存。

即使管道如此强大，在公有的API中却不常见。例如，我梳理过Go的标准库，在145个包中有超过6000个公有的API。在这上千个API中，去重后，只有5个用到了管道。

在公有的API中使用管道时，如何折衷考虑和取舍，缺乏指导。“共有API”，我是指“任何实现者和使用者是不同的两个人的编程接口”。这篇文章会深入讲解，为如何在共有API中使用管道，提供一系列的原则和解释。一些特例会在本章末尾讨论。

原则 #1

API应该声明管道的方向性。

例子

time.After

func After(d Duration) <-chan Time

signal.Notify

func Notify(c chan<- os.Signal, sig ...os.Signal)

尽管并不常用，Go允许指定一个管道的方向性。语言规范这么写：

可选的<-操作符指定了管道的方向，发送或接收。如果没有指定方向，那么管道就是双向的。

关键在于API签名中的方向操作符会被编译器强制检查

t := time.After(time.Second)  t <- time.Now()  // 会编译失败(send to receive-only type <-chan Time)

除了能够被编译器强制检查安全性，方向操作符还能帮助API使用者理解数据的流动方向——只需要看一下类型签名即可。

原则 #2

向一个管道发送无界数据流的API必须写文档解释清楚在消费者消费不及时时API的行为。

例子

time.NewTicker

// NewTicker returns a new Ticker containing a channel that will send the  // time with a period specified by the duration argument.  // It adjusts the intervals or drops ticks to make up for slow receivers.  // ...  func NewTicker(d Duration) *Ticker {      ...  }

signal.Notify

// Notify causes package signal to relay incoming signals to c.  // ...  // Package signal will not block sending to c  // ...    func Notify(c chan<- os.Signal, sig ...os.Signal) {

ssh.Conn.OpenChannel

// OpenChannel tries to open an channel.  // ...  // On success it returns the SSH Channel and a Go channel for  // incoming, out-of-band requests. The Go channel must be serviced, or  // the connection will hang.    OpenChannel(name string, data []byte) (Channel, <-chan *Request, error)

当一个API向一个管道发送无界数据流时，在实现API时面临的问题是如果向管道发送数据会阻塞怎么办。阻塞的原因可能是管道已经满了或者管道是无缓冲的，没有go协程准备好接收数据。针对不同的场景要选择合适的行为，但是每个场景必须作出选择。例如，ssh包选择了阻塞，并且文档写明如果你不接受数据，连接就会被卡住。signal.Notify 和 time.Tick选择不阻塞，直接丢弃数据。

不足的是，Go本身并没有从类型或函数签名角度提供方法指定默认行为。作为API的设计者，你必须在文档中写明行为，不然其行为就是不定的。然而，多数情况下我们都是API的使用者而不是设计者，所以我们可以反过来记这个原则，反过来就是一条警告信息：

对于通过一个管道向一个慢速的消费者发送无界数据的API，在没有通读API的文档或者实现源码之前，你不能确定API的行为。

原则 #3

向一个管道发送有界数据，同时这个管道是作为参数传递进来的API，必须用文档写明对于慢速消费者的行为。

不好的例子

rpc.Client.Go

func (client *Client) Go(serviceMethod string,                           args interface{},                           reply interface{},                           done chan *Call                           ) *Call

这个原则和第二个原则类似，不同点在于这个原则用于发送有界数据的API。不幸的是，在标准库中没有很好的例子。标准库中唯一的API就是rpc.Client.Go，但它违背了我们的原则。文档上这么写：

Go异步的调用这个函数。它会返回代表着调用的Call数据结构。在调用完成时，done管道会通过返回同一个Call对象来触发。如果done是空的，Go会分配一个新的管道；如果不空，done必须是有缓冲的，不然Go就会崩溃。

Go发送了有界数据（只有1，当远程调用结束时）。但是注意到，由于管道是被当作参数传递到函数中的，所以它仍然存在慢速消费者问题。即使你必须传一个带缓冲的管道进来，如果管道已满，向这个管道发送数据仍然可能会阻塞。文档并没有定义这种场景下的行为。需要我们来读读源码了：

src/pkg/net/rpc/client.go

 func (call *Call) done() {       select {       case call.Done <- call:       // ok       default:       // We don't want to block here.  It is the caller's responsibility to make           // sure the channel has enough buffer space. See comment in Go().           if debugLog {               log.Println("rpc: discarding Call reply due to insufficient Done chan capacity")           }       }  }

噢！如果done管道没有合适的缓冲，RPC的响应可能丢失了。

原则 #4

向一个管道发送无界数据流的API应该接受管道作为参数，而不是返回一个新的管道。

例子

signal.Notify

func Notify(c chan<- os.Signal, sig ...os.Signal)

ssh.NewClient

func NewClient(c Conn, chans <-chan NewChannel, reqs <-chan *Request) *Client

当我第一次看到signal.Notify这个API时，我很疑惑，“为什么它接收一个管道作为输入而不是直接返回一个管道给我用？”“使用这个API需要调用方分配一个管道，难道API就不能替我们做么，像下面这样？”

func Notify(sig ...os.Signal) <-chan os.Signal

文档帮助我们理解为什么这不是好的选择：

signal包向c发送数据时并不会阻塞：调用方必须保证c有足够的缓冲空间来跟得上潜在的信号速度

signal.Notify接收管道作为参数，因为它把缓冲空间的控制权交给了调用方。这使得调用方可以选择，在处理一个信号时，可以安全的忽略多少信号，这需要和缓存这些信号的内存开销作折衷考虑。

缓冲大小的控制在高吞吐系统中尤为重要。设想一个高吞吐的发布订阅系统的这样一个接口：

func Subscribe(topic string, msgs chan<- Msg)

往管道中发送越多的消息，管道同步称为性能瓶颈的可能性越大。由于API允许调用方创建管道，调用方需要考虑缓冲，进而性能可以由调用方控制。这是一种更灵活的设计。

如果仅仅是控制缓冲的大小，我们可能会争论如下的API就足够了：

func Notify(sig ...os.Signal, bufSize int) <-chan os.Signal

这样设计，管道作为参数还是必须的，因为这样允许调用方使用一个管道动态的处理不同类型的信号。这样设计为调用方提供了更多的程序结构和性能上的灵活性。作为一个假想实验，让我们用Subscribe API来构建需求。订阅newcustomer管道，并对于每一条消息，为消费者订阅其主题。如果API允许我们传递接收管道，我们可以这样写：

msgs := make(chan Msg, 128)    Subscribe("newcustomer", msgs)  for m := range msgs {      switch m.Topic {      case "newcustomer":          Subscribe(msg.Payload, msgs)      default:          handleCustomerMessage(m)  }

但是，如果管道被返回了，调用方不得不为每一个订阅启动一个单独的go协程。这在任何复用场景都会带来额外的内存和同步开销:

for m := range Subscribe("newcustomer") {      go subCustomer(m.Payload)  }    func subCustomer(topic string) {      for m := range Subscribe(topic) {          handleCustomerMessage(m)      }  }

原则 #5

发送有界数据的API可以通过返回一个合适大小缓冲的管道来达到目的。

例子:

http.CloseNotifier

type CloseNotifier interface {          // CloseNotify returns a channel that receives a single value          // when the client connection has gone away.          CloseNotify() <-chan bool  }

time.After

func After(d Duration) <-chan Time

当API向一个管道发送有界数据时，可以返回一个拥有容纳全部数据的缓冲空间的管道。这个要返回的管道的方向性标识保证了调用方必须遵守约定。CloseNotify 和After返回的管道 都利用了这一点。

同时，需要注意到，通过允许调用方传递一个管道来接收数据，这些调用可能会更灵活。但需要处理当管道满了的时候（原则3）。例如，另外一个可选的，更灵活的CloseNotifier：

type CloseNotifier interface {          // CloseNotify sends a single value with the ResponseWriter whose          // underlying connection has gone away.          CloseNotify(chan<- http.ResponseWriter)  }

但是这种额外的灵活性带来的开销并不值得关注，因为单一的调用方很少会同时等待多个关闭通知。毕竟，关闭通知只有在某个连接上下文内才有效。不同的连接一般都是相互独立的。

特例

一些API打破了我们的原则，需要仔细分析。

原则 #1 的特例

API需要声明管的方向性。

例子

rpc.Client.Go

传过来的done管道没有方向性标识符：

func (client *Client) Go(serviceMethod string,                           args interface{},                           reply interface{},                           done chan *Call                           ) *Call

直观上看，这样做是因为done管道是作为Call结构体的一部分返回的。

type Call struct {          // ...          Done          chan *Call  // Strobes when call is complete.  }

这种灵活性是需要的，这样允许在你传nil时分配一个done管道出来。如果坚持原则1，就需要从Call结构中去除done并且声明两个函数：

func (c *Client) Go(method string,                      args interface{},                      reply interface{}                      ) (*Call, <-chan *Call)    func (c *Client) GoEx(method string,                        args interface{},                        reply interface{},                        done chan<- *Call                        ) *Call

原则 #4 的特例

向管道发送无界数据流的API需要接收管道作为参数，而不是返回一个新的管道。

例子

go.crypto/ssh

func NewClientConn(c net.Conn, addr string, config *ClientConfig)      (Conn, <-chan NewChannel, <-chan *Request, error)

time.Tick

func Tick(d Duration) <-chan Time

go.crypto/ssh包几乎在所有的地方都返回了无界的数据流管道。ssh.NewClientConn只是其中的一个。给调用者更多控制权和灵活性的API应该是这样：

func NewClientConn(c net.Conn,                     addr string,                     config *ClientConfig,                     channels chan<- NewChannel,                     reqs chan<- *Request                     ) (Conn, error)

time.Tick也违反了这个原则，但是易于理解。我们很少会创建非常多的计时器，通常都是独立的处理不同的计时器。这个例子中缓冲也没太大意义。

第二部分：那些原本可能使用的管道

这篇文章是一篇长文，所以我准备分成两部分讲。接下来会提很多问题，为什么标准库中可以使用管的地方却没有用管道。例如，http.Serve 返回了一个永不结束的等待被处理的请求流，为什么用了回调函数而不是将这些请求发送到一个处理管道中？第二部分会介绍更多！

原文链接： Alan Shreve   翻译： 伯乐在线 - Codefor
译文链接： http://blog.jobbole.com/73700/