原文在此。遗憾的是文章只提出了问题，并没明确提供如何解决这些问题。但无论如何，对于这种可以引起反思的文章，是不能放过的。另外，我得承认，似乎高层次的分布式系统的抽象，用函数式语言的范式来表述更容易一些（实现上其实未必）。

————翻译分隔线————

channel 独木难支
或者说为什么流水线作业没那么容易

勇敢和聪明的 Golang 并发模型。

@kachayev 撰写

概述

Go 被设计用于更容易的构建并发系统，因此它有运行独立的计算的 goroutine 和用于它们之间通讯的 channel。我们之前都听过这个故事。所有的例子和指南看起来都挺好的：我们可以创建一个新的 channel，可以向这个 channel 发送数据，可以从 channel 读取，甚至还有漂亮和优雅的 select 语句（顺便提一下，为什么 21 世纪了我们还在用语句？），阻塞读和缓存……

主旨：99% 的情况下，我其实并不关心响应是由 channel 传递的，还是一只魔法独角兽从它的角上带来的。

在为初学者撰写指南的时候这确实挺酷的！但是当你尝试实现大型的复杂系统的时候这就很痛苦了。channel 太原始了。它们是低级的构件，我相当怀疑你愿意在日常工作中天天和它们打交道。

看看“高级模式”和“流水作业”。不是那么简单吧？有太多的东西要考虑，并且永远记得：什么时候、如何关闭 channel；如何传递错误；如何释放资源。我抱怨这些是因为我曾尝试实现一些东西，然后失败了。而我每天都在面对这些东西。

你可能会说，对于初学者没必要理解所有细节。不过……描述一个模式真得很“高级”？不幸的是，答案是否定的。它们是基础和常识。

更仔细的了解一下流水作业问题。这真得是流水作业？不，“…对于每个来自目录的路径计算 MD5 校验码，并将结果存入一个 map[string][string]…”。这只是一个 pmap（并行 map）。或具有池化执行器的、有限的并行化的 pmap。而 pmap 不应当需要我输入如此多行代码。想了解真正的流水作业——我将在文章的最后介绍一个（参阅“构建 Twitter 分析器”的段落）。

那么模式如何呢？

为了快速开发真实的应用，我们应当能够提炼出比原始的 channel 层面更高的抽象。它们只是传输层。我们需要应用层的抽象来编写程序（对比 OSI），否则你会发现你总是在低级的 channel 网络的细节上纠结，试图在生产环境中、偶发的、没有任何有效的方法重现的找到它不工作的原因。参阅 Erlang OTP 是如何针对性的解决类似的问题的：将你保护在低级的消息传递代码之外。

低级的代码有什么问题？这里有一篇超棒的文章“爱德华 C++ 手（译注：借‘爱德华剪刀手’）”：

手里有把剪刀并不一定总是那么糟糕。爱德华有许多天赋：例如，他能创造劲爆的狗狗的发型。别误会——它展示了许多劲爆的狗狗的发型（我是说优雅且简单的 C++ 代码），但是主要内容还是关于如何避免剪坏，以及在发生剪坏的情况下进行急救。

在 Kyiv Go 聚会的时候，我经历了相同的情况：在一页幻灯上那 20 行干净可读的代码。一个不一般的竞态条件和一个可能出现的运行时错误。这对于所有听众来说很明显吗？不。至少一半人不明白。

痛苦的缘由？

好，让我们试着收集一些类似的模式。从工作的经验中、从书中、从其他语言中（是，伙计们，我知道这有点令人难以相信，不过还有许多其他语言同样也有并发的设计）。

Rob Pike 讨论了 Fan-in、Fan-out。在许多情况下，这很有用，不过还是关于网络的 channel。而不是你的应用。在任何情况下，看看（无耻的从这里偷的）。
Rob Pike talks about Fan-in, Fan-out. It’s useful in many ways, but still about the network of channels. Not about your application. In any case, let’s check (shamelessly stolen from here).

func merge(cs ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)

    // 为 cs 中每个输入的 channel 启动一个输出用的 goroutine。
    // 从 c 中复制值出来直到 c 被关闭，然后又调用 wg.Done。
    output := func(c <-chan int) {
        for n := range c {
            out <- n
        }
        wg.Done()
    }
    wg.Add(len(cs))
    for _, c := range cs {
        go output(c)
    }

    // 一旦所有输出的 goroutine 完成的，就启动一个 goroutine 来关闭 out。
    // 这必须在 wg.Add 调用后启动。
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

呃…… <-chan int。在我的应用中重用起来没那么抽象（例如，迁移到库中）……并且在每次我需要的时候都重新实现也不是那么清晰。那么如何让其可以重用？<-chan interface{}？欢迎来到类型转换和运行时错误的领地。如果，希望实现一个高级的 fan-in（合并）就必须牺牲类型安全。同样的（不幸的）是其他模式也是一样。

我真正想要的是：

func merge[T](cs ...<-chan T) <-chan T

是，我知道 Go 没有泛型，因为谁需要它们呢？

现在天气如何？

回到模式。让我们分析一个假设的项目，服务器端开发会与实际经验非常接近。我们需要一个服务器接收请求，输入一个美国的州，返回从 OpenWeatherMap 收集到的信息。例如这样：

$ http localhost:4912/weather?q=CA
HTTP/1.1 200 OK
Access-Control-Allow-Credentials: true
Access-Control-Allow-Methods: GET, POST
Access-Control-Allow-Origin: *
Connection: keep-alive
Content-Type: application/json; charset=utf-8

[{
    "clouds": {
        "all": 40
    },
    "id": 5391959,
    "main": {
        "temp": 288.89,
        "temp_max": 291.48,
        "temp_min": 286.15
    },
    "name": "San Francisco",
    "weather": [
        {
            "description": "mist",
            "icon": "50d",
            "id": 701,
            "main": "Mist"
        }
    ]
}, {
    "clouds": {
        "all": 90
    },
    "id": 5368361,
    "main": {
        "temp": 292.83,
        "temp_max": 296.15,
        "temp_min": 289.15
    },
    "name": "Los Angeles",
    "weather": [
        {
            "description": "mist",
            "icon": "50d",
            "id": 701,
            "main": "Mist"
        }
    ]
}]

pmap

让我们从一些我们已经知道的东西开始。那么，我们收到了请求 ?q=CA。我不想对从哪里得到相关城市的列表进行解释。我们可以用这个数据库，在内存中缓存以及其他什么合理的东西。假设我们有一个神奇的 findCities(state) 函数，返回 chan City（像通常 go 程序表现的延迟序列那样）。然后呢？每个城市我们都必须调用 OpenWeatherMap API 并解析结果到一个 map[City]Weather 中。我们已经讨论过这个模式了。这是个 pmap。我希望我的代码像这样：

chanCities := findCities(state)
resolver := func(name City) Weather { return openWeatherMap.AskFor(name) }
weather := chanCities.Par.Map(resolver)

或限制并发数：

chanCities := findCities(state)
pool := NewWorkers(20)
resolver := func(w Worker, name City) Weather { return w.AskFor(name) }
weather := chanCities.Par.BoundedMap(pool, resolver)

我希望所有这些 <-done 同步和神圣的 select 完全被隐藏起来。

Futures & Promises

获取当前天气可能需要很长的时间，例如，你有一个很长的城市列表。当然，你不希望重复的 API 调用，因此应当可以用某种方法管理并行的请求：

func collect(state string) Weather {
  calc, ok := calculations.get(state) // check if it's in progress
  if !ok {
      calc.calculations.run(state) // run otherwise
  }
  return calc.Wait() // wait until done
}

这也被叫做 future/promise 。Wiki 的解释：

它们描述了一个对象对于结果扮演了代理的角色，而这在一开始是不可预知的，通常是由于它的值尚未完成计算造成的。

我已经听过太多人说 go 的 future 很简单：

f := make(chan int, 1)

这是错误的，因为所有的等待者都应当得到结果（译注：实现 channel 的订阅与变化值的广播确实另人头大）。而这个版本也是错的：

[C]
f := make(chan int, 1)
v <- f f <- v // 在这里使用 v [/C] 由于不可能用这个方法管理资源。所以，当某个家伙在他的代码里丢了 f <- v 那部分，我希望你能幸运的发现这个 bug。

将数据直接发给所有的等待者来实现 promise 没那么复杂（我不确定这段代码是不是有 bug）：

type PromiseDelivery chan interface{}
type Promise struct {
    sync.RWMutex
    value interface{}
    waiters []PromiseDelivery
}

func (p *Promise) Deliver(value interface{}) {
    p.Lock()
    defer p.Unlock()
    p.value = value
    for _, w := range p.waiters {
        locW := w
        go func(){
            locW <- value
        }()
    }
}

func (p *Promise) Value() interface{} {
    if p.value != nil {
        return p.value
    }

    delivery := make(PromiseDelivery)
    p.waiters = append(p.waiters, delivery)
    return <-delivery
}

func NewPromise() *Promise {
    return &Promise{
        value: nil,
        waiters: []PromiseDelivery{},
    }
}

如何使用他呢？

p := NewPromise()
go func(){
  p.Deliver(42)
}()
p.Value().(int) // 阻塞，当有值的时候返回 interface{} 

不过这里有 interface{} 和类型转换。我实际上想要什么呢？

// 在那些经过良好测试的库，甚至 stdlib 中
type PromiseDelivery[T] chan T
type Promise[T] struct {
    sync.RWMutex
    value T
    waiters []PromiseDelivery[T]
}
func (p *Promise[T]) Deliver(value T)
func (p *Promise[T]) Value() T
func NewPromise[T]() *Promise[T]

// 我的代码:
v := NewPromise[int]()
go func(){
  v.Deliver("woooow!") // 错误
  v.Deliver(42)
}()
v.Value() // 阻塞并返回 42，而不是 interface{}

是的，当然，没有人需要泛型。我在讨论什么鬼玩意啊？

也可以通过使用 select 来避免 p.Lock() 来监听 deliver，并在一个 goroutine 中 wait 操作。还可以引入对最终用户极为有用的 .ValueWithTimeout 方法。还有许多许多其他“你可以……”。尽管我们实际上是在讨论一个 20 行的代码（它的长度可能在每次你发现 future/promise 交互更多细节的时候就开始增长了）。我真得需要知道（或想到） channel 为我传递值吗？不！

pub/sub

假设我们想要构建一个实时服务。那么我们的客户端现在可以开启一个 websocket 连接，传递 q=CA 请求，并即刻获得加利福尼亚的天气变化情况。它看起来应该像：

// deliverer
calculation.WhenDone(func(state string, w Weather) {
  broker.Publish("CA", w)
})

// client
ch := broker.Subscribe("CA")
for update := range ch {
  w.Write(update.Serialize())
}

这是一个典型的 pub/sub（译注：公告/订阅）。你可以从高级 Go 模式的演讲中学习它，甚至可以找到即刻可用的实现。问题是，它们全都基于接口的。

有没有可能实现：

broker := NewBroker[String, Weather]()
// so that
broker.Subs(42) // compilation failure
// and
broker.Subs("CA") // returns (chan Weather) not (chan interface{})

当然！如果你能勇敢的在项目之间复制粘贴代码，并到处进行修改。

map/filter

假设希望给与我们的用户更多的弹性，从而引入了新的查询参数：show，它的值可以是 all|temp|wind|icon。

可能你可以从基础开始：

ch := broker.Subscribe("CA")
for update := range ch {
  temps := []Temp
  for _, t := update.Temp {
    temps = append(temps, t)
  }

  w.Write(temps)
}

不过，在写了 10 个这样的方法之后，你会意识到它没那么模块化，并且也很无聊。可能你需要：

ch := broker.Subscribe("CA").Map(func(w Weather) Temp { return w.Temp })
for update := range ch {
  w.Write(update)
}

等等，我有提过 channel 是一个 functor（译注：函子）吗？跟 future/promise 一样。

p := NewPromise().Map(func(w Weather) Temp { return w.Temp })
go func(){
  p.Deliver(Weather{Temp{42}})
}()
p.Value().(Temp) // Temp, not Weather

这意味着我重用了 future 的 channel 的相同代码。你也可以用想 transducers 这样的东西来完成它。我经常在 ClojureScript 的代码中使用的技巧：

(->> (send url) ;; returns chan, put single value to it {:status 200 :result 42} when ready
     (async/filter< #(= 200 (:status %))) ;; check that :status is 200
     (async/map< :result)) ;; expose only 42 to end user
;; note, that it will close all channels (including implicit intermediate one) properly

当我可以简单的进行 x.Map(transformation) 并得到相同类型的值的时候，我真得需要关心 x 是个 channel 还是个 future 吗？在这个例子里，为什么允许我创建 make(chan int) 而不能创建 make(Future int) 呢？

Request/Reply

假设我们的用户喜欢这个服务，并且频繁的使用它。那么就需要引入一些简单的 API 限制：每天、每个 IP 请求的数量。收集这个数量保存在一个 map[string]int 中很简单。Go 的文档说“不要通过共享内存来通讯，用通讯来共享内存”。好吧，听起来是个好主意。

req := make(chan string)
go func() { // wow, look here - it's an actor!
  m := map[string]int{}
  for r := range req {
    if v, ok := m[r]; !ok {
      m[r] = 1
    } else {
      m[r] = v + 1
    }
  } 
}()

go func() {
  req <- "127.0.0.2"
}()

go func() {
  req <- "127.0.0.1"
}()

这很容易。现在可以计算每个 IP 请求的数量了。不但如此……同时也可以要求执行请求需要权限。

type Req struct {
  ip string
  resp chan int
}

func NewRequest(ip string) *Req {
  return &Req{ip, make(chan int)}
}

requests := make(chan *Req)

go func() {
  m := map[string]int{}
  for r := range requests {
    if v, ok := m[r.ip]; !ok {
      m[r.ip] = 1
    } else {
      m[r.ip] = v + 1
    }
    r.resp <- m[r.ip]
  } 
}()

go func() {
  r := NewRequest("127.0.0.2")
  requests <- r
  fmt.Println(<- r.resp)
}()

go func() {
  r := NewRequest("127.0.0.1")
  requests <- r
  fmt.Println(<- r.resp)
}()

我不会再问你要泛型的解决方案（没有写死的 string 和 int）。换而言之，我希望你检查一下这段代码中是不是都正确？真得这么简单吗？

你确定 r.resp <- m[r.ip] 是个好办法？不，肯定不是！我希望有任何人等待那些很慢的客户端。是吗？而如果我有许多很慢的客户端的时候会怎么样呢？可能我需要对此进行一些处理。

而 requests <- r 这部分简单吗？如果我的 actor（服务器）过载无法响应的时候呢？可能我需要在这里处理超时……

时不时的我就需要特定的初始化和清理过程……都需要超时机制。并且需要能保持请求，直到初始化完成。

那么调用的优先权呢？例如，当我需要为了分析系统实现 Dump 方法，但是又不想让所有的用户暂停来收集需要的数据。

还有……看看 Erlang 中的 gen_server。为了保险起见，我希望它一被实现就可以是具有良好的文档的，经过高度覆盖测试的库。98% 的情况下，我不希望看到这样的介绍：make(chan int, ?) 而我不想思考到底我应该将 ? 替换成多少。

99% 的情况下，我其实并不关心响应是由 channel 传递的，还是一只魔法独角兽从它的角上带来的。

数不胜数

还有许多其他常见的并发的情况。我想你已经明白了。

苦难

你可以说，这些模式都不常见。不过……我在我的项目中不得不实现它们中的大多数。每！一！次！可能我不怎么走运，而你的项目会跟写给初学者的指南一样简单。

我知道，你们中的大多数会说“世界是艰辛的，编程是苦难的”。我会继续打击你：至少有一些语言展示了部分解决这些问题的示例。至少，在尝试解决它。Haskell 和 Scala 的类型系统提供了构建强大的高级抽象的能力，甚至自定义控制流来处理并发。而另一阵营的 Clojure 利用动态类型鼓励和共享高级的抽象。Rust 有 channel 和泛型。

让它工作 -> 让它优雅 -> 让它可重用。

现在，第一步已经完成。接下来呢？不要误会，go 是一个有远见的语言：channel 和 goroutine 比起例如 pthread 来说更好，不过是不是真得就停留在此？

补充：构建 Twitter 分析器

关于真实的流水作业。

你可能已经看过 Twitter 的分析了，它真得很棒。假设它尚未出现，而我们需要自己的分析工具：提供一个用户名，来统计有多少用户看过（至少是理论上）他的 tweet。应该如何做呢？其实不难：读取用户的时间轴，过滤掉所有的 retweet 和回复，然后请求其他 tweet 的 retweeter，为每个 retweeter 请求 follower 的列表，合并所有 retweeter 的 follower 在一起，然后加上这个用户的 follower。对于这个步骤我想要的结果是：
map[TweetId][]Username（retweeter）和 map[Username][]Username。这些用于构造一个向请求者展示的魔幻的表格是足够了。

有一些技术细节你应当留意：

Twitter API 需要每个调用都使用 OAuth，并且设定了很强的限制（每个用户每 15 分钟 450 次调用）。为了对付这个限制，我们将用预定义的一个 OAuth token 列表（例如 50 个）组织在一个池中供 worker 使用，每个 worker 在达到限制之前都可以让自己休息一会。
大多数 Twitter API 调用通过 since_id 或 max_id 使用了结果分页。因此你不能依赖一个请求就可以获取完整的结果。

一个粗糙的实现的例子。注意，你没必要理解这个文件中所有的内容。相反，如果你无法理解的话，这恰恰说明我们做对了。

那么我们现在有什么？

• 一些步骤的计算：TimelineReader -> RetweetersReader -> FollowersReader -> FinalReducer。
• 自供消息。由于分页所有阶段都是递归的。这意味着每个步骤都会向下一个阶段和其本身发出消息。在这个情况下，很难处理取消的情况。甚至无法发现某个步骤的工作全部完成。
• 尽早传播。至少有两种情况：首先为了通过 TweetId 来收集 []Username，我们需要将收集到的信息直接从 RetweetersReader 发送到 FinalReducer。然后，一开始我们就知道，需要获得初始用户的 follower，因此他的用户名应当从 TimelineReader 传递到
  RetweetersReader 步骤。
• 中间收缩。FollowersReader 不只是一个管道。它会过滤我们已经见过的用户名（因为总不想重复工作吧）。
• 持续工作的 worker。在许多情况下，你无法等待 worker 退出。例如，当你实现了一个服务，它会同时响应许多客户端的时候。

---