如何优雅地实现并发编排任务

前端 2023-07-05 17:29:38

126阅读

文中转载微信公众平台「吴亲强的深夜厨房」，创作者吴亲杜兰特。转截文中请联络吴亲强的深夜厨房微信公众号。

业务场景

在做任务开发设计的情况下，大家一定会遇到下列情景:

情景1：启用第三方接口的情况下，一个要求你需要启用不一样的插口，做数据拼装。

情景2:一个运用主页很有可能借助于许多服务项目。那么就牵涉到在载入网页页面时必须另外要求好几个服务项目的插口。这一步通常是由后端开发统一启用拼装数据信息再回到给前面，也就是说白了的 BFF(Backend For Frontend) 层。

对于之上二种情景，假定在沒有强相互依赖下，挑选串行通信启用，那麼总用时即:

 
  time=s1 s2 ....sn

依照当今秒入上百万的有为青年，那么长期早已将你祖宗十八代问好了一遍。

为了更好地杰出的KPI，大家通常会挑选高并发地启用这种依靠插口。那麼总用时便是:

 
  time=max(s1,s2,s3.....,sn)

自然逐渐堆业务流程的情况下能够先串行化，直到上边的人心急的情况下，展示绝技。

那样，年末 PPT 就可以再加上浓厚的一笔流水账单:为业务流程某一插口提升百分XXX特性，间接性造成XXX使用价值。

自然这一切的前提条件是，做老总不明白技术性，做技术性”懂”你。

大破冲霄楼,假如改动成高并发启用，你很有可能会那么写，

 
  package main 
   
  import ( 
      "fmt" 
      "sync" 
      "time" 
  ) 
   
  func main() { 
      var wg sync.WaitGroup 
      wg.Add(2) 
   
      var userInfo *User 
      var productList []Product 
   
      go func() { 
          defer wg.Done() 
          userInfo, _ = getUser() 
      }() 
   
      go func() { 
          defer wg.Done() 
          productList, _ = getProductList() 
      }() 
      wg.Wait() 
      fmt.Printf("客户信息:% v\n", userInfo) 
      fmt.Printf("产品信息:% v\n", productList) 
  } 
   
   
  /********客户服务**********/ 
   
  type User struct { 
      Name string 
      Age uint8 
  } 
   
  func getUser() (*User, error) { 
      time.Sleep(500 * time.Millisecond) 
      var u User 
      u.Name = "wuqinqiang" 
      u.Age = 18 
      return &u, nil 
  } 
   
  /********产品服务项目**********/ 
   
  type Product struct { 
      Title string 
      Price uint32 
  } 
   
  func getProductList() ([]Product, error) { 
      time.Sleep(400 * time.Millisecond) 
      var list []Product 
      list = append(list, Product{ 
          Title: "SHib", 
          Price: 10, 
      }) 
      return list, nil 
  }

从完成上而言，必须是多少服务项目，会开多少个 G，运用 sync.WaitGroup 的特点，

完成高并发编辑每日任务的实际效果。

仿佛，问题不大。

可是伴随着编号 996 业务场景的提升，你能发觉，许多控制模块都是有类似的作用，仅仅相匹配的业务场景不一样罢了。

那麼大家能否抽像出一套对于此业务场景的专用工具，而把实际业务流程完成交到业务流程方。

应用

秉着不反复造轮子的标准，去搜过下开源软件，最后看到了 go-zero 里边的一个专用工具 mapreduce。

能够自主 Google 这一专有名词。

应用非常简单。大家根据它更新改造一下上边的编码:

 
  package main 
   
  import ( 
      "fmt" 
      "github.com/tal-tech/go-zero/core/mr" 
      "time" 
  ) 
   
  func main() { 
      var userInfo *User 
      var productList []Product 
      _ = mr.Finish(func() (err error) { 
          userInfo, err = getUser() 
          return err 
      }, func() (err error) { 
          productList, err = getProductList() 
          return err 
      }) 
      fmt.Printf("客户信息:% v\n", userInfo) 
      fmt.Printf("产品信息:% v\n", productList) 
  } 
  //打印出 
  客户信息:&{Name:wuqinqiang Age:18} 
  产品信息:[{Title:SHib Price:10}]

是否舒适多了。

可是这儿还必须留意一点，假定你启用的在其中一个服务项目不正确，而且你 return err 相匹配的不正确，那麼别的启用的服务项目会被撤销。

例如大家改动 getProductList 立即回应不正确。

 
  func getProductList() ([]Product, error) { 
      return nil, errors.New("test error") 
  } 
  //打印出 
  // 客户信息:<nil> 
  // 产品信息:[]

那麼最后打印出的情况下连客户信息都是会为空，由于发生一个服务项目不正确，客户服务要求被取消了。

一般状况下，在要求服务项目不正确的情况下大家会出现最低实际操作，一个服务项目不正确不可以危害别的要求的結果。

因此在应用的情况下实际解决在于业务场景。

源代码

即然用了，那麼就追下源码吧。

 
  func Finish(fns ...func() error) error { 
      if len(fns) == 0 { 
          return nil 
      } 
   
      return MapReduceVoid(func(source chan<- interface{}) { 
          for _, fn := range fns { 
              source <- fn 
          } 
      }, func(item interface{}, writer Writer, cancel func(error)) { 
          fn := item.(func() error) 
          if err := fn(); err != nil { 
              cancel(err) 
          } 
      }, func(pipe <-chan interface{}, cancel func(error)) { 
          drain(pipe) 
      }, WithWorkers(len(fns))) 
  }

 
  func MapReduceVoid(generator GenerateFunc, mapper MapperFunc, reducer VoidReducerFunc, opts ...Option) error { 
      _, err := MapReduce(generator, mapper, func(input <-chan interface{}, writer Writer, cancel func(error)) { 
          reducer(input, cancel) 
          drain(input) 
          // We need to write a placeholder to let MapReduce to continue on reducer done, 
          // otherwise, all goroutines are waiting. The placeholder will be discarded by MapReduce. 
          writer.Write(lang.Placeholder) 
      }, opts...) 
      return err 
  }

针对 MapReduceVoid涵数，关键查询三个闭包主要参数。

第一个 GenerateFunc 用以生产制造数据信息。
MapperFunc 载入生产制造出的数据信息，开展解决。
VoidReducerFunc 这儿表明不对 mapper 后的数据信息做汇聚回到。因此这一闭包在这里实际操作基本上0功效。

 
  func MapReduce(generate GenerateFunc, mapper MapperFunc, reducer ReducerFunc, opts ...Option) (interface{}, error) { 
      source := buildSource(generate)  
      return MapReduceWithSource(source, mapper, reducer, opts...) 
  } 
   
  func buildSource(generate GenerateFunc) chan interface{} { 
      source := make(chan interface{})// 建立无缓存安全通道 
      threading.GoSafe(func() { 
          defer close(source) 
          generate(source) //逐渐生产制造数据信息 
      }) 
   
      return source //回到无缓存安全通道 
  }

buildSource涵数中，回到一个无缓存的安全通道。并打开一个 G 运作 generate(source)，往无缓存安全通道塞数据信息。这一generate(source) 不便是一开始 Finish 传送的第一个闭包主要参数。

 
  return MapReduceVoid(func(source chan<- interface{}) { 
      // 就这个 
          for _, fn := range fns { 
              source <- fn 
          } 
      })

随后查询 MapReduceWithSource 涵数,

 
  func MapReduceWithSource(source <-chan interface{}, mapper MapperFunc, reducer ReducerFunc, 
      opts ...Option) (interface{}, error) { 
      options := buildOptions(opts...) 
      //每日任务实行完毕通告数据信号 
      output := make(chan interface{}) 
      //将mapper解决完的数据信息载入collector 
      collector := make(chan interface{}, options.workers) 
      // 撤销实际操作数据信号 
      done := syncx.NewDoneChan() 
      writer := newGuardedWriter(output, done.Done()) 
      var closeOnce sync.Once 
      var retErr errorx.AtomicError 
      finish := func() { 
          closeOnce.Do(func() { 
              done.Close() 
              close(output) 
          }) 
      } 
      cancel := once(func(err error) { 
          if err != nil { 
              retErr.Set(err) 
          } else { 
              retErr.Set(ErrCancelWithNil) 
          } 
   
          drain(source) 
          finish() 
      }) 
   
      go func() { 
          defer func() { 
              if r := recover(); r != nil { 
                  cancel(fmt.Errorf("%v", r)) 
              } else { 
                  finish() 
              } 
          }() 
          reducer(collector, writer, cancel) 
          drain(collector) 
      }() 
      // 真真正正从制作器安全通道取数据信息实行Mapper 
      go executeMappers(func(item interface{}, w Writer) { 
          mapper(item, w, cancel) 
      }, source, collector, done.Done(), options.workers) 
   
      value, ok := <-output 
      if err := retErr.Load(); err != nil { 
          return nil, err 
      } else if ok { 
          return value, nil 
      } else { 
          return nil, ErrReduceNoOutput 
      } 
  }

这一段编码挺长的，大家说下关键的点。这儿应用一个G 启用 executeMappers 方式。

 
  go executeMappers(func(item interface{}, w Writer) { 
          mapper(item, w, cancel) 
      }, source, collector, done.Done(), options.workers)

 
  func executeMappers(mapper MapFunc, input <-chan interface{}, collector chan<- interface{}, 
      done <-chan lang.PlaceholderType, workers int) { 
      var wg sync.WaitGroup 
      defer func() { 
          // 等候全部每日任务所有实行结束 
          wg.Wait() 
          // 关掉安全通道 
          close(collector) 
      }() 
     //依据特定总数建立 worker池 
      pool := make(chan lang.PlaceholderType, workers)  
      writer := newGuardedWriter(collector, done) 
      for { 
          select { 
          case <-done: 
              return 
          case pool <- lang.Placeholder: 
              // 从buildSource() 回到的无缓存安全通道取数据信息 
              item, ok := <-input  
              // 当安全通道关掉，完毕 
              if !ok { 
                  <-pool 
                  return 
              } 
   
              wg.Add(1) 
              // better to safely run caller defined method 
              threading.GoSafe(func() { 
                  defer func() { 
                      wg.Done() 
                      <-pool 
                  }() 
                  //真真正正运作闭包涵数的地区 
                 // func(item interface{}, w Writer) { 
                 // mapper(item, w, cancel) 
                 // } 
                  mapper(item, writer) 
              }) 
          } 
      } 
  }

实际的逻辑性已备注名称，编码非常容易懂。

一旦 executeMappers 涵数回到，关掉 collector 安全通道，那麼实行 reducer 不会再堵塞。

 
  go func() { 
          defer func() { 
              if r := recover(); r != nil { 
                  cancel(fmt.Errorf("%v", r)) 
              } else { 
                  finish() 
              } 
          }() 
          reducer(collector, writer, cancel) 
          //这儿 
          drain(collector) 
      }()

这儿的 reducer(collector, writer, cancel) 实际上就是以 MapReduceVoid 传送的第三个闭包涵数。

 
  func MapReduceVoid(generator GenerateFunc, mapper MapperFunc, reducer VoidReducerFunc, opts ...Option) error { 
      _, err := MapReduce(generator, mapper, func(input <-chan interface{}, writer Writer, cancel func(error)) { 
          reducer(input, cancel) 
          //这儿 
          drain(input) 
          // We need to write a placeholder to let MapReduce to continue on reducer done, 
          // otherwise, all goroutines are waiting. The placeholder will be discarded by MapReduce. 
          writer.Write(lang.Placeholder) 
      }, opts...) 
      return err 
  }

随后这一闭包涵数又实行了 reducer(input, cancel)，这儿的 reducer 便是大家一开始表述过的 VoidReducerFunc，从 Finish() 而成。

这些,见到上边三个地区的 drain(input)了没有?

 
  // drain drains the channel. 
  func drain(channel <-chan interface{}) { 
      // drain the channel 
      for range channel { 
      } 
  }

实际上便是一个排尽 channel 的实际操作，可是三个地区都对同一个 channel做一样的实际操作，也是要我难以相信。

也有更关键的一点。

 
  go func() { 
          defer func() { 
              if r := recover(); r != nil { 
                  cancel(fmt.Errorf("%v", r)) 
              } else { 
                  finish() 
              } 
          }() 
          reducer(collector, writer, cancel) 
          drain(collector) 
      }()

上边的编码，倘若实行 reducer，writer 载入引起 panic,那麼drain(collector) 将沒有机遇实行。

但是创作者早已修补了这个问题，立即把 drain(collector) 放进到 defer。

实际 issues[1]。

到这儿，有关 Finish 的源代码也就结束了。有兴趣的能够看一下别的源代码。

很喜欢 go-zero 里的一些专用工具，可是专用工具通常并不单独，取决于别的文件包，造成本来只为应用在其中一个专用工具却必须安裝全部包。

因此最后的結果便是扒源代码，建立无依靠库工具箱，遵照 MIT 就可以。

附则[1]https://github.com/tal-tech/go-zero/issues/676

the end

免责声明：本文不代表本站的观点和立场，如有侵权请联系本站删除！本站仅提供信息存储空间服务。

如何优雅地实现并发编排任务

业务场景

应用

源代码

精选推荐

让 Flutter 在鸿蒙系统上跑起来

小冰一口气发布11个AI歌手：仅训练45天媲美专业级歌手

京东智能客服品牌焕新：“言犀”亮相2020京东JDD大会

AI改进建筑施工安全的十种方式

六项任务、多种数据类型，谷歌、DeepMind提出高效Transformer评估基准

第四范式NeurIPS 2020：知识图谱嵌入的自动化

随机推荐

如何优雅地实现并发编排任务

业务场景

应用

源代码

精选推荐

让 Flutter 在鸿蒙系统上跑起来

小冰一口气发布11个AI歌手：仅训练45天 媲美专业级歌手

京东智能客服品牌焕新：“言犀”亮相2020京东JDD大会

AI改进建筑施工安全的十种方式

六项任务、多种数据类型，谷歌、DeepMind提出高效Transformer评估基准

第四范式NeurIPS 2020：知识图谱嵌入的自动化

随机推荐

小冰一口气发布11个AI歌手：仅训练45天媲美专业级歌手