Go并发实践

废话不多说，先来几行代检验下你是否适合本文,如果你发现看不懂建议先去看看简单点的东西。

go f() go f("abc", 123) ch := make(chan int) go func() { c <- 123}() fmt.Println(<-ch) 

简单的例子

ok，下面假设这样一个场景，有一家新闻媒体会持续向官方网站输出最新消息，刚好他们的后端提供了一个api可以获取指定分类的最新消息以及该类别预计下次有新消息的时间。我们再假设一下，他们还提供了一个SDK来帮助我们封装好了api：

func FetchNews(kind string) (newsList []News, next time.Time, err error)  type News struct {  Kind  string  Title string  Text  string } 

翻看了一下SDK之后，发现还有这样俩玩意：

type Fetcher interface {  Fetch() (newsList []News, next time.Time, err error) }  func Fetch(kind string) Fetcher  

现在希望有个这样一个channel，我们可以一直从中读到news，剩下的逻辑交给channel的另一头来搞定。同时我们还希望之歌channel中可以读到多个不同分类的news。下面先来定义一下消费者

type Subscription interface {  // 返回一个可以读到news的channel  Flow() <-chan News  // 用来关闭channel  Close() error } // 用来将Fetcher转换成消费者 func Subscribe(fetcher Fetcher) Subscription // 用来合并多个channel的消息 func Merge(subs ...Subscription) Subscription 

现在把代码主体结构搭一下

merged := Merge(  Subscribe(Fetch("体育")),  Subscribe(Fetch("财经")),  Subscribe(Fetch("房产")), )  // 假设只需要保持3秒，然后关闭 time.AfterFunc(time.Second*3, func() {  fmt.Println("close msg:", merged.Close()) })  for news := range merged.Flow() {  fmt.Println(news.Title) } 

接下来定义个sub结构体来实现一下Subscription这个接口，同时用它来完成Subscribe方法

type sub struct {  fetcher Fetcher  flow    chan News }  // sub的主体逻辑，持续网channel里放数据 func (s *sub) loop() { }  // 返回channel func (s *sub) Flow() <-chan News {  return s.flow }  // 用来关闭channel func (s *sub) Close() error {  return nil }  // 用来将Fetcher转换成消费者 func Subscribe(fetcher Fetcher) Subscription {  s := &sub{   fetcher: fetcher,   flow:    make(chan News),  }  go s.loop()  return s } 

loop怎么搞

那么问题来了，既然主要干活的是loop，那么loop需要做什么呢？首先肯定需要调用Fetch方法来拿消息，然后将拿到的消息放进channel去，然后别忘了响应Close方法调用，即退出循环。来试着写下loop

// sub的主体逻辑，持续网channel里放数据 func (s *sub) loop() {  // 首先大部分时候肯定是死循环读消息了  for {   // 既然是死循环，那么肯定要有个地方判断是不是该关闭了，给s加个closed字段   if s.closed {    close(s.flow)    return   }   // 用它的fetcher来拿数据   newsList, next, err := s.fetcher.Fetch()   if err != nil {    // 错误也要存下来，在Close的时候返回，那再给s加个err字段吧    s.err = err    time.Sleep(time.Second * 3)    continue   }   // 把拿到的数据都丢进channel   for _, news := range newsList {    s.flow <- news   }   // 既然提供了下次有消息的事件，那就在到点之前"睡"一会吧   if now := time.Now(); next.After(now) {    time.Sleep(next.Sub(now))   }  } } 

这一波操作之后，sub结构体就变成了这样：

type sub struct {  fetcher Fetcher  flow    chan News  closed  bool  // 是否已经关了  err     error // 存放出现的错误 } 

顺便把Close也可以写出来了：

// 用来关闭channel func (s *sub) Close() error {  s.closed = true  return s.err } 

好像少点什么

大部分都完成了，然后需要把上面的坑填一下，首先给SDK来点假数据

func FetchNews(kind string) (newsList []News, next time.Time, err error) {  newsList = append(newsList, News{   Kind:  kind,   Title: time.Now().Format("2006-01-02"),   Text:  time.Now().Format("15:04:05"),  })  next = time.Now().Add(time.Second * 5)  return } 

SDK里还有个fetcher

type Fetcher interface {  Fetch() (newsList []News, next time.Time, err error) }  type fet struct {  kind string }  func (f fet) Fetch() (newsList []News, next time.Time, err error) {  return FetchNews(f.kind) }  func Fetch(kind string) Fetcher {  return fet{kind: kind} } 

还有一个merge

// 用来合并多个channel的消息 func Merge(subs ...Subscription) Subscription {  res := &sub{   flow: make(chan News),  }  for _, s := range subs {   go func(s Subscription) {    res.flow <- <-s.Flow()   }(s)  }  return res } 

到这就算”完成”了，代码总是有bug的，来找找bug吧

bug

来看看sub的closed跟err字段，它在loop及Close方法中都用到了，而且这俩方法应该是在不同的goroutine。这样就有了并发安全问题。

if s.closed {}  s.closed = true 

另外这俩time.Sleep也不太妙，假如在他“睡”的时候调用了Close呢，岂不是要等它醒了才能关掉

time.Sleep(time.Second * 3)  time.Sleep(next.Sub(now)) 

再一个比较难发现，就是有可能会导致假死的地方，仔细看看这句

s.flow <- news 

我们知道channel都是阻塞的(杠精别拿缓冲区说事，想让这里保证不阻塞，你说缓冲区需要多大？)，如果我在调用了Close之后这行代码碰巧执行了，那么实际上它会永远卡在这里。找到这仨问题之后，来借助select一次性搞定这仨bug

改bug

select是个神奇的东西，它就像站在常年堵车的十字路口的交警，让可以通行的车辆先走（这个比喻起始不太恰当，堵车是互相堵，channel阻塞多半跟其他channel无关）。现在请select出场

// sub的主体逻辑，持续网channel里放数据 func (s *sub) loop() {  // 将这三个遍历定义在外面复用  var err error  var next time.Time  var newsList []News   // 做个缓冲区，可以让读写独立工作  var cache []News   // 首先大部分时候肯定是死循环读消息了  for {   // 首次执行的时候，delay为0   // 非首次执行时，计算下次请求数据是什么时候   var delay time.Duration   if now := time.Now(); next.After(now) {    delay = next.Sub(now)   }   // 这里可以将之前的`time.Sleep`做成一个通道，到时会从通道读出超时的时刻   chFetch := time.After(delay)    select {   // 循环控制改成了用channel控制，在Close不调用的情况下，这个写操作会一直阻塞，select就会执行其他case   case s.closeCh <- err:    // 进这个case说明closeCh可写了，也就是说它的读操作正在发生，也就是说Close被调用了    close(s.flow)    return   case <-chFetch:    // 进这个case说明`time.After`时间到了，该执行下一次fetch了    newsList, next, err = s.fetcher.Fetch()    if err != nil {     // 原来的3秒后重试，可以调整为3秒后再触发下一次fetch，效果完全一样     next = time.Now().Add(time.Second * 3)     break // 需要break掉当前case    }    // 将数据放进缓冲区，然后就不管后续逻辑了，读写分离    cache = append(cache, newsList...)   case s.flow <- cache[0]:    // 进这个case说明成功“消费”一个news，把它从cache中踢走，然后开始下一个循环    cache = cache[1:]   }  } } 

相应的sub结构体与Close方法：

type sub struct {  fetcher Fetcher  flow    chan News  closeCh chan error // 把error放进channel，跟关闭用同一个 }  // 用来关闭channel func (s *sub) Close() error {  // 从这里读出error并返回，其他时候是不会从中读数据的，而不读数据就会让写操作阻塞  return <-s.closeCh } 

到这还有点问题，如果cache里没数据了怎么办，继续改

向值为nil的channel发数据

给for循环里加个局部变量ch，每次循环一开始这个ch就变成了nil，当cache有数据就紧接着给ch赋值。下面只放了有改动的代码

这里有个知识点，对值为nil的channel执行读写操作不会panic，只会block

for {     // 每次循环ch都是nil  var ch chan News  // 也定义个news变量，防止select里面空指针  var news News  if len(cache) > 0 {   // cache有数据时才给ch赋值   ch = s.flow   news = cache[0]  }  select {     // ch为空则不会进这个case，ch不为空才会执行逻辑  case ch <- news:   // 进这个case说明成功“消费”一个news，把它从cache中踢走，然后开始下一个循环   cache = cache[1:]  } } 

到这里其实还有可优化的地方，比如这个cache现在理论上可以无限“膨胀”，现在来给它加个限制

限制缓冲区大小

没什么好啰嗦的，直接上代码

// 来个变量定义缓冲区最大值 const cacheSize = 10  chFetch := time.After(delay) // 在这加个逻辑，如果缓冲区满了，就给这个ch置为nil，让下面阻塞，就不会有新数据了 if len(cache) >= cacheSize {  chFetch = nil } 

结语

代码优化总是无止境的，这里也只是相对优化而已。代码中还有一些地方可以精雕细琢，本文主要关注并发这块东西，所以其他的欢迎大家在评论区嗨所欲嗨。