使用 go 协程+Channel,让你的代码执行快到起飞

  作者: horryhuang,腾讯 PCG 后台开发工程师

  传统的串行代码执行,逻辑比较简单,当数据量比较大时,执行效率低下,既然我们使用 go,那就利用 go 相对与其他语言的优势,轻量化的协程以及 channel,接下来让我们使用 go 协程+chan,让我们的代码速度来个大的飞跃吧~

  背景:最近做了一个需求,是产品小姐姐提的对于流失用户的召回,精简一下流程,首先从表 1 中取出符合多少天未登陆条件的用户 uid,然后利用这个用户 uid 在表 2 中进行比较(如果用户曾经被召回过,会在表 2 中留下一条记录,之后就不再召回),如果表 2 中有该用户的记录,就不做任何操作,如果没有,则触发用户召回的服务。当然实际业务比这个要复杂,但只从这个精简的业务中,也能找到很多优化我们代码的地方,从而提高效率。

  第一次尝试 demo:

  在接到这个需求的时候,心情非常开心,这不就是我 sql boy 发挥作用的时候了吗?于是,很快我就撸出了代码。大致的 demo 如下(实际业务中不要写魔法数字):

  var uidTargetList []int

  var uidList []int

  var id int

  for {

  // 每次从表1中取出100个用户,这里id用户遍历,每次取出数据后,返回最后一个用户记录对应的id,然后使用这个id作为读表的比较条件,防止取出重复用户

  if uidList, id, err :=lastLoginTimeStore.GetUnloginUserByPage(id,

  100, startTime, endTime); err !=nil {

  rlog.Error("get unlogin user by page err", rlog.Err(err))

  }

  if len(uidList)==0 {

  break

  }

  for index :=range uidList {

  var hasSent bool

  // 判断用户是否被召回过,如果没有,则加入uidTargetList,以便后续触发召回服务

  if hasSent, err :=callbackStore.HasSent(uidList[index]); err !=nil {

  rlog.Error("get user record error", rlog.Int("uid", uidList[index]), rlog.Err(err))

  }

  if !hasSent {

  uidTargetList=append(uidTargetList, uidList[index])

  }

  }

  }

  然后我就和产品说,我写好了,服务可以跑了,当天产品就要我先灰度发送,我就发了 16w 用户,正当我服务跑起来准备刷刷 km 时,我发现这个速度也太慢了,大概每分钟居然只能处理 600 个用户,照着这个速度,还不得发到明天,产品可能要把我打死,于是我马上终止了服务,马上进行优化。

  第一次优化:

  马上我就仔细分析这个服务的瓶颈在哪,这个服务中有 2 次与数据库的交互,这种操作一般就是效率低的缘由。这里的第一次 io 操作从表 1 中取出用户数据,每次取出 100 条记录,如果增加每次取出的数据,可能会带来超时的风险,同时这样的效率提升也比较小,没有量级的提升,很明显,这个 io 操作不是我优化的主要目标。于是我将目标放到了第二个 io 操作,每次只能比较一个用户,这样的效率比较低,所以,我应该优化这个地方,如果我能和第一次 io 操作一样,能够每次比较 100 个用户,这样的提升就是量级了,想到这里,我瞬间又重新寻回了新手程序员的蜜汁自信。

  那怎样才能一下比较很多个用户数据,马上,我就想起了可以使用协程啊,有一个用户的数据,就 go 一个协程去比较,这样的效率不就得到了极大的提升,然后我有一次撸起了袖子,又开始干了。这次代码的 demo 感觉就比第一版高端了许多。主要是利用 uidChan 和 uidTargetChan 在多协程中传递数据,uidChan 传递从表 1 中查询出的数据,然后在表 2 中比较,如果符合条件,则将其存入 uidTargetChan,最后再利用 uidTargetList 这个切片,存放所有符合条件的用户 uid。

  // uidHandler 创建一个结构体,包括一个等待队列,然后uidChan 用于在多个协程中传递用户uid

  type uidHandler struct {

  wg sync.WaitGroup

  uidChan chan int

  }

  // uidTargetHandler 同样的这个结构体包括一个等待队列,然后uidTargetChan 用于在多个协程中传递符合条件的用户uid

  type uidTargetHandler struct {

  wg sync.WaitGroup

  uidTargetChan chan int

  }

  func test1() {

  uh :=uidHandler{

  wg: sync.WaitGroup{},

  uidChan: make(chan int, 100),

  }

  uth :=uidTargetHandler{

  wg: sync.WaitGroup{},

  uidTargetChan: make(chan int, 100),

  }

  // 利用协程启动获取targetUid的服务

  go func() {

  getTargetUid(uh, uth)

  }()

  // 记录下这些targetUid,uidTargetList就是最后保存所有符合条件的uid

  var uidTargetList []int

  go func() {

  RecordTargetUid(uth, &uidTargetList)

  }()

  var uidList []int

  var id int

  for {

  // 每次从表1中取出100个用户,这里id用户遍历,每次取出数据后,返回最后一个用户记录对应的id,然后使用这个id作为读表的比较条件,防止取出重复用户

  if uidList, id, err=lastLoginTimeStore.GetUnloginUserByPage(id,

  100, startTime, endTime); err !=nil {

  rlog.Error("get unlogin user by page err", rlog.Err(err))

  }

  if len(uidList)==0 {

  break

  }

  // 将取出的uid直接放入uh.uidChan

  for index :=range uidList {

  uh.uidChan

  uh.wg.Add(1)

  }

  }

  uh.wg.Wait()

  uth.wg.Wait()

  // 当走到这一步时,所有的目标用户的uid全部保存在 uidTargetList 中了

  }

  然后我们来看看 getTargetUid 和 RecordTargetUid 的代码:

  // getTargetUid 获取目标uid,即可以发送通知的用户

  func getTargetUid(uh uidHandler, uth uidTargetHandler) {

  for {

  uid :=

  uh.wg.Done()

  // 对于用户的uid,直接并发去比较,如果符合条件,就放入uth.uidTargetChan

  go func(userUid int) {

  var hasSent bool

  var err error

  if hasSent, err=callbackStore.HasSent(userUid); err !=nil {

  rlog.Error("get user record error", rlog.Int("uid", userUid), rlog.Err(err))

  }

  if !hasSent {

  uth.uidTargetChan

  uth.wg.Add(1)

  }

  }(uid)

  }

  }

  // RecordTargetUid 记录下可以发送用户的uid,实际业务中应该是直接利用这些uid去启动后续服务

  func RecordTargetUid(uth uidTargetHandler, uidTargetList *[]int) {

  for {

  uid :=

  *uidTargetList=append(*uidTargetList, uid)

  uth.wg.Done()

  }

  }

  至此,我们就能将所有符合条件的用户 uid 放在 uidTargetList。然后我想着,这样性能就有了量的提升,产品小姐姐待会要夸我真快,真给力。然后我就重启了服务。但。。。,猝不及防的事情又发生了,报了这个“use of closed network connection”错误,经过分析,可能是我协程开了太多了,一下子并发了太多协程去和数据库交互,然后导致出错,进而连接被关闭,最终报了这个错。于是,想着能不能不要并发那么多协程,对同时跑的协程数量进行一个限制。所以又想到了线程池,可以仿造这个概念弄个协程池,但是谷歌了一下,线程池主要就是节省线程的创建和销毁的时间,但是对于协程而言,它的创建和销毁本来就消耗不大,go 的协程本来就是非常轻量的,go 开发中一般也不建议使用线程池。然后我又陷入了深思,代码好难,人生也好难。

  第二次优化:

  自己的脑瓜不够转了,只能去求助外援。然后我只能去请教了 dayo 大哥,然后 dayo 传授了我一个江湖典藏小诀窍,专治这个毛病。即利用 for 循环,只开启固定的协程去处理这些用户 uid,在服务器可以承载的范围,这样就不会有特别多的协程同时与数据库交互了。利用这个诀窍,我对 getTargetUid 函数进行了小小的修改,就解决了这个问题,getTargetUid 修改后的代码如下:

  // getTargetUid 获取目标uid,即可以发送通知的用户

  func getTargetUid(uh uidHandler, uth uidTargetHandler) {

  // 只并发100个协程,然后这些协程循环去从chan中读取并进行相应的处理

  for i :=0; i < 100; i++ {

  go func() {

  for {

  uid :=

  uh.wg.Done()

  var hasSent bool

  var err error

  if hasSent, err=callbackStore.HasSent(uid); err !=nil {

  rlog.Error("get user record error", rlog.Int("uid", uid), rlog.Err(err))

  }

  if !hasSent {

  uth.uidTargetChan

  uth.wg.Add(1)

  }

  }

  }()

  }

  }

  这次,服务又跑起来了,大概每分钟 8000 个用户,速度大大提升,产品小姐姐知道了我的壮举后,对我赞不绝口,菜鸡程序员的快乐又有了,这就是我利用 go 协程提升了服务的效率,总的来说,go 的 chan 非常好用,很方便在多协程间传递数据,chan+协程简直就是利器,还在用线程池的 java 同学听到了都羡慕哭了。

  结语

  当然这只是优化的一部分,比如你的表中用户记录一共有 2 亿条,这样依次遍历效率仍然太低了,可以将用户数据分段,比如每 100 万个数据分为一段,每一段 go 一个协程去处理,这样读取的效率也有了极大的提升,还可以增加多台服务器等等,这些都可以提升速度,但这些就不是本文的重点啦,大家可以自己试着用多协程+chan 去优化一下自己的代码,提升代码的运行速度吧~

上一篇:Gin使用自定义中间件--JWT


下一篇:28. scp远程拷贝