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

 [[399159]]

本文转载自微信公众号「吴亲强的深夜食堂」，作者吴亲库里。转载本文请联系吴亲强的深夜食堂公众号。

成都创新互联公司是一家集成都网站设计、网站建设、网站页面设计、网站优化SEO优化为一体的专业网站设计公司，已为成都等多地近百家企业提供网站建设服务。追求良好的浏览体验,以探求精品塑造与理念升华,设计最适合用户的网站页面。 合作只是第一步，服务才是根本,我们始终坚持讲诚信，负责任的原则，为您进行细心、贴心、认真的服务,与众多客户在蓬勃发展的市场环境中,互促共生。

业务场景

在做任务开发的时候，你们一定会碰到以下场景:

场景1：调用第三方接口的时候， 一个需求你需要调用不同的接口，做数据组装。

场景2:一个应用首页可能依托于很多服务。那就涉及到在加载页面时需要同时请求多个服务的接口。这一步往往是由后端统一调用组装数据再返回给前端，也就是所谓的 BFF(Backend For Frontend) 层。

针对以上两种场景，假设在没有强依赖关系下，选择串行调用，那么总耗时即:

 
 
 
 
    
  
  
  
  
  1. time=s1+s2+....sn 
    2.

按照当代秒入百万的有为青年，这么长时间早就把你祖宗十八代问候了一遍。

为了伟大的KPI，我们往往会选择并发地调用这些依赖接口。那么总耗时就是:

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

当然开始堆业务的时候可以先串行化，等到上面的人着急的时候，亮出绝招。

这样，年底 PPT 就可以加上浓重的一笔流水账:为业务某个接口提高百分之XXX性能，间接产生XXX价值。

当然这一切的前提是，做老板不懂技术，做技术”懂”你。

言归正传,如果修改成并发调用，你可能会这么写，

 
 
 
 
    
  
  
  
  
  1. package main 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3. import ( 
    4.   
  
  
  
  4.     "fmt" 
    5.   
  
  
  
  5.     "sync" 
    6.   
  
  
  
  6.     "time" 
    7.   
  
  
  
  7. ) 
    8.   
  
  
  
  8.  
    9.   
  
  
  
  9. func main() { 
    10.   
  
  
  
  10.     var wg sync.WaitGroup 
    11.   
  
  
  
  11.     wg.Add(2) 
    12.   
  
  
  
  12.  
    13.   
  
  
  
  13.     var userInfo *User 
    14.   
  
  
  
  14.     var productList []Product 
    15.   
  
  
  
  15.  
    16.   
  
  
  
  16.     go func() { 
    17.   
  
  
  
  17.         defer wg.Done() 
    18.   
  
  
  
  18.         userInfo, _ = getUser() 
    19.   
  
  
  
  19.     }() 
    20.   
  
  
  
  20.  
    21.   
  
  
  
  21.     go func() { 
    22.   
  
  
  
  22.         defer wg.Done() 
    23.   
  
  
  
  23.         productList, _ = getProductList() 
    24.   
  
  
  
  24.     }() 
    25.   
  
  
  
  25.     wg.Wait() 
    26.   
  
  
  
  26.     fmt.Printf("用户信息:%+v\n", userInfo) 
    27.   
  
  
  
  27.     fmt.Printf("商品信息:%+v\n", productList) 
    28.   
  
  
  
  28. } 
    29.   
  
  
  
  29.  
    30.   
  
  
  
  30.  
    31.   
  
  
  
  31. /********用户服务**********/ 
    32.   
  
  
  
  32.  
    33.   
  
  
  
  33. type User struct { 
    34.   
  
  
  
  34.     Name string 
    35.   
  
  
  
  35.     Age uint8 
    36.   
  
  
  
  36. } 
    37.   
  
  
  
  37.  
    38.   
  
  
  
  38. func getUser() (*User, error) { 
    39.   
  
  
  
  39.     time.Sleep(500 * time.Millisecond) 
    40.   
  
  
  
  40.     var u User 
    41.   
  
  
  
  41.     u.Name = "wuqinqiang" 
    42.   
  
  
  
  42.     u.Age = 18 
    43.   
  
  
  
  43.     return &u, nil 
    44.   
  
  
  
  44. } 
    45.   
  
  
  
  45.  
    46.   
  
  
  
  46. /********商品服务**********/ 
    47.   
  
  
  
  47.  
    48.   
  
  
  
  48. type Product struct { 
    49.   
  
  
  
  49.     Title string 
    50.   
  
  
  
  50.     Price uint32 
    51.   
  
  
  
  51. } 
    52.   
  
  
  
  52.  
    53.   
  
  
  
  53. func getProductList() ([]Product, error) { 
    54.   
  
  
  
  54.     time.Sleep(400 * time.Millisecond) 
    55.   
  
  
  
  55.     var list []Product 
    56.   
  
  
  
  56.     list = append(list, Product{ 
    57.   
  
  
  
  57.         Title: "SHib", 
    58.   
  
  
  
  58.         Price: 10, 
    59.   
  
  
  
  59.     }) 
    60.   
  
  
  
  60.     return list, nil 
    61.   
  
  
  
  61. } 
    62.

从实现上来说，需要多少服务，会开多少个 G，利用 sync.WaitGroup 的特性，

实现并发编排任务的效果。

好像，问题不大。

但是随着代号 996 业务场景的增加，你会发现，好多模块都有相似的功能，只是对应的业务场景不同而已。

那么我们能不能抽像出一套针对此业务场景的工具，而把具体业务实现交给业务方。

使用

本着不重复造轮子的原则，去搜了下开源项目，最终看上了 go-zero 里面的一个工具 mapreduce。

可以自行 Google 这个名词。

使用很简单。我们通过它改造一下上面的代码:

 
 
 
 
    
  
  
  
  
  1. package main 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3. import ( 
    4.   
  
  
  
  4.     "fmt" 
    5.   
  
  
  
  5.     "github.com/tal-tech/go-zero/core/mr" 
    6.   
  
  
  
  6.     "time" 
    7.   
  
  
  
  7. ) 
    8.   
  
  
  
  8.  
    9.   
  
  
  
  9. func main() { 
    10.   
  
  
  
  10.     var userInfo *User 
    11.   
  
  
  
  11.     var productList []Product 
    12.   
  
  
  
  12.     _ = mr.Finish(func() (err error) { 
    13.   
  
  
  
  13.         userInfo, err = getUser() 
    14.   
  
  
  
  14.         return err 
    15.   
  
  
  
  15.     }, func() (err error) { 
    16.   
  
  
  
  16.         productList, err = getProductList() 
    17.   
  
  
  
  17.         return err 
    18.   
  
  
  
  18.     }) 
    19.   
  
  
  
  19.     fmt.Printf("用户信息:%+v\n", userInfo) 
    20.   
  
  
  
  20.     fmt.Printf("商品信息:%+v\n", productList) 
    21.   
  
  
  
  21. } 
    22.   
  
  
  
  22. //打印 
    23.   
  
  
  
  23. 用户信息:&{Name:wuqinqiang Age:18} 
    24.   
  
  
  
  24. 商品信息:[{Title:SHib Price:10}] 
    25.

是不是舒服多了。

但是这里还需要注意一点，假设你调用的其中一个服务错误，并且你 return err 对应的错误，那么其他调用的服务会被取消。

比如我们修改 getProductList 直接响应错误。

 
 
 
 
    
  
  
  
  
  1. func getProductList() ([]Product, error) { 
    2.   
  
  
  
  2.     return nil, errors.New("test error") 
    3.   
  
  
  
  3. } 
    4.   
  
  
  
  4. //打印 
    5.   
  
  
  
  5. // 用户信息: 
    6.   
  
  
  
  6. // 商品信息:[] 
    7.

那么最终打印的时候连用户信息都会为空，因为出现一个服务错误，用户服务请求被取消了。

一般情况下，在请求服务错误的时候我们会有保底操作，一个服务错误不能影响其他请求的结果。

所以在使用的时候具体处理取决于业务场景。

源码

既然用了，那么就追下源码吧。

 
 
 
 
    
  
  
  
  
  1. func Finish(fns ...func() error) error { 
    2.   
  
  
  
  2.     if len(fns) == 0 { 
    3.   
  
  
  
  3.         return nil 
    4.   
  
  
  
  4.     } 
    5.   
  
  
  
  5.  
    6.   
  
  
  
  6.     return MapReduceVoid(func(source chan<- interface{}) { 
    7.   
  
  
  
  7.         for _, fn := range fns { 
    8.   
  
  
  
  8.             source <- fn 
    9.   
  
  
  
  9.         } 
    10.   
  
  
  
  10.     }, func(item interface{}, writer Writer, cancel func(error)) { 
    11.   
  
  
  
  11.         fn := item.(func() error) 
    12.   
  
  
  
  12.         if err := fn(); err != nil { 
    13.   
  
  
  
  13.             cancel(err) 
    14.   
  
  
  
  14.         } 
    15.   
  
  
  
  15.     }, func(pipe <-chan interface{}, cancel func(error)) { 
    16.   
  
  
  
  16.         drain(pipe) 
    17.   
  
  
  
  17.     }, WithWorkers(len(fns))) 
    18.   
  
  
  
  18. }  
    19.  
 
 
 
 
 
 
 
    
  
  
  
  
  1. func MapReduceVoid(generator GenerateFunc, mapper MapperFunc, reducer VoidReducerFunc, opts ...Option) error { 
    2.   
  
  
  
  2.     _, err := MapReduce(generator, mapper, func(input <-chan interface{}, writer Writer, cancel func(error)) { 
    3.   
  
  
  
  3.         reducer(input, cancel) 
    4.   
  
  
  
  4.         drain(input) 
    5.   
  
  
  
  5.         // We need to write a placeholder to let MapReduce to continue on reducer done, 
    6.   
  
  
  
  6.         // otherwise, all goroutines are waiting. The placeholder will be discarded by MapReduce. 
    7.   
  
  
  
  7.         writer.Write(lang.Placeholder) 
    8.   
  
  
  
  8.     }, opts...) 
    9.   
  
  
  
  9.     return err 
    10.   
  
  
  
  10. } 
    11.

对于 MapReduceVoid函数，主要查看三个闭包参数。

  • 第一个 GenerateFunc 用于生产数据。
  • MapperFunc 读取生产出的数据，进行处理。
  • VoidReducerFunc 这里表示不对 mapper 后的数据做聚合返回。所以这个闭包在此操作几乎0作用。
 
 
 
 
    
  
  
  
  
  1. func MapReduce(generate GenerateFunc, mapper MapperFunc, reducer ReducerFunc, opts ...Option) (interface{}, error) { 
    2.   
  
  
  
  2.     source := buildSource(generate)  
    3.   
  
  
  
  3.     return MapReduceWithSource(source, mapper, reducer, opts...) 
    4.   
  
  
  
  4. } 
    5.   
  
  
  
  5.  
    6.   
  
  
  
  6. func buildSource(generate GenerateFunc) chan interface{} { 
    7.   
  
  
  
  7.     source := make(chan interface{})// 创建无缓冲通道 
    8.   
  
  
  
  8.     threading.GoSafe(func() { 
    9.   
  
  
  
  9.         defer close(source) 
    10.   
  
  
  
  10.         generate(source) //开始生产数据 
    11.   
  
  
  
  11.     }) 
    12.   
  
  
  
  12.  
    13.   
  
  
  
  13.     return source //返回无缓冲通道 
    14.   
  
  
  
  14. } 
    15.

buildSource函数中，返回一个无缓冲的通道。并开启一个 G 运行 generate(source)，往无缓冲通道塞数据。这个generate(source) 不就是一开始 Finish 传递的第一个闭包参数。

 
 
 
 
    
  
  
  
  
  1. return MapReduceVoid(func(source chan<- interface{}) { 
    2.   
  
  
  
  2.     // 就这个 
    3.   
  
  
  
  3.         for _, fn := range fns { 
    4.   
  
  
  
  4.             source <- fn 
    5.   
  
  
  
  5.         } 
    6.   
  
  
  
  6.     }) 
    7.

然后查看 MapReduceWithSource 函数,

 
 
 
 
    
  
  
  
  
  1. func MapReduceWithSource(source <-chan interface{}, mapper MapperFunc, reducer ReducerFunc, 
    2.   
  
  
  
  2.     opts ...Option) (interface{}, error) { 
    3.   
  
  
  
  3.     options := buildOptions(opts...) 
    4.   
  
  
  
  4.     //任务执行结束通知信号 
    5.   
  
  
  
  5.     output := make(chan interface{}) 
    6.   
  
  
  
  6.     //将mapper处理完的数据写入collector 
    7.   
  
  
  
  7.     collector := make(chan interface{}, options.workers) 
    8.   
  
  
  
  8.     // 取消操作信号 
    9.   
  
  
  
  9.     done := syncx.NewDoneChan() 
    10.   
  
  
  
  10.     writer := newGuardedWriter(output, done.Done()) 
    11.   
  
  
  
  11.     var closeOnce sync.Once 
    12.   
  
  
  
  12.     var retErr errorx.AtomicError 
    13.   
  
  
  
  13.     finish := func() { 
    14.   
  
  
  
  14.         closeOnce.Do(func() { 
    15.   
  
  
  
  15.             done.Close() 
    16.   
  
  
  
  16.             close(output) 
    17.   
  
  
  
  17.         }) 
    18.   
  
  
  
  18.     } 
    19.   
  
  
  
  19.     cancel := once(func(err error) { 
    20.   
  
  
  
  20.         if err != nil { 
    21.   
  
  
  
  21.             retErr.Set(err) 
    22.   
  
  
  
  22.         } else { 
    23.   
  
  
  
  23.             retErr.Set(ErrCancelWithNil) 
    24.   
  
  
  
  24.         } 
    25.   
  
  
  
  25.  
    26.   
  
  
  
  26.         drain(source) 
    27.   
  
  
  
  27.         finish() 
    28.   
  
  
  
  28.     }) 
    29.   
  
  
  
  29.  
    30.   
  
  
  
  30.     go func() { 
    31.   
  
  
  
  31.         defer func() { 
    32.   
  
  
  
  32.             if r := recover(); r != nil { 
    33.   
  
  
  
  33.                 cancel(fmt.Errorf("%v", r)) 
    34.   
  
  
  
  34.             } else { 
    35.   
  
  
  
  35.                 finish() 
    36.   
  
  
  
  36.             } 
    37.   
  
  
  
  37.         }() 
    38.   
  
  
  
  38.         reducer(collector, writer, cancel) 
    39.   
  
  
  
  39.         drain(collector) 
    40.   
  
  
  
  40.     }() 
    41.   
  
  
  
  41.     // 真正从生成器通道取数据执行Mapper 
    42.   
  
  
  
  42.     go executeMappers(func(item interface{}, w Writer) { 
    43.   
  
  
  
  43.         mapper(item, w, cancel) 
    44.   
  
  
  
  44.     }, source, collector, done.Done(), options.workers) 
    45.   
  
  
  
  45.  
    46.   
  
  
  
  46.     value, ok := <-output 
    47.   
  
  
  
  47.     if err := retErr.Load(); err != nil { 
    48.   
  
  
  
  48.         return nil, err 
    49.   
  
  
  
  49.     } else if ok { 
    50.   
  
  
  
  50.         return value, nil 
    51.   
  
  
  
  51.     } else { 
    52.   
  
  
  
  52.         return nil, ErrReduceNoOutput 
    53.   
  
  
  
  53.     } 
    54.   
  
  
  
  54. } 
    55.

这段代码挺长的，我们说下核心的点。这里使用一个G 调用 executeMappers 方法。

 
 
 
 
    
  
  
  
  
  1. go executeMappers(func(item interface{}, w Writer) { 
    2.   
  
  
  
  2.         mapper(item, w, cancel) 
    3.   
  
  
  
  3.     }, source, collector, done.Done(), options.workers) 
    4.  
 
 
 
 
 
 
 
    
  
  
  
  
  1. func executeMappers(mapper MapFunc, input <-chan interface{}, collector chan<- interface{}, 
    2.   
  
  
  
  2.     done <-chan lang.PlaceholderType, workers int) { 
    3.   
  
  
  
  3.     var wg sync.WaitGroup 
    4.   
  
  
  
  4.     defer func() { 
    5.   
  
  
  
  5.         // 等待所有任务全部执行完毕 
    6.   
  
  
  
  6.         wg.Wait() 
    7.   
  
  
  
  7.         // 关闭通道 
    8.   
  
  
  
  8.         close(collector) 
    9.   
  
  
  
  9.     }() 
    10.   
  
  
  
  10.    //根据指定数量创建 worker池 
    11.   
  
  
  
  11.     pool := make(chan lang.PlaceholderType, workers)  
    12.   
  
  
  
  12.     writer := newGuardedWriter(collector, done) 
    13.   
  
  
  
  13.     for { 
    14.   
  
  
  
  14.         select { 
    15.   
  
  
  
  15.         case <-done: 
    16.   
  
  
  
  16.             return 
    17.   
  
  
  
  17.         case pool <- lang.Placeholder: 
    18.   
  
  
  
  18.             // 从buildSource() 返回的无缓冲通道取数据 
    19.   
  
  
  
  19.             item, ok := <-input  
    20.   
  
  
  
  20.             // 当通道关闭，结束 
    21.   
  
  
  
  21.             if !ok { 
    22.   
  
  
  
  22.                 <-pool 
    23.   
  
  
  
  23.                 return 
    24.   
  
  
  
  24.             } 
    25.   
  
  
  
  25.  
    26.   
  
  
  
  26.             wg.Add(1) 
    27.   
  
  
  
  27.             // better to safely run caller defined method 
    28.   
  
  
  
  28.             threading.GoSafe(func() { 
    29.   
  
  
  
  29.                 defer func() { 
    30.   
  
  
  
  30.                     wg.Done() 
    31.   
  
  
  
  31.                     <-pool 
    32.   
  
  
  
  32.                 }() 
    33.   
  
  
  
  33.                 //真正运行闭包函数的地方 
    34.   
  
  
  
  34.                // func(item interface{}, w Writer) { 
    35.   
  
  
  
  35.                // mapper(item, w, cancel) 
    36.   
  
  
  
  36.                // } 
    37.   
  
  
  
  37.                 mapper(item, writer) 
    38.   
  
  
  
  38.             }) 
    39.   
  
  
  
  39.         } 
    40.   
  
  
  
  40.     } 
    41.   
  
  
  
  41. } 
    42.

具体的逻辑已备注，代码很容易懂。

一旦 executeMappers 函数返回，关闭 collector 通道，那么执行 reducer 不再阻塞。

 
 
 
 
    
  
  
  
  
  1. go func() { 
    2.   
  
  
  
  2.         defer func() { 
    3.   
  
  
  
  3.             if r := recover(); r != nil { 
    4.   
  
  
  
  4.                 cancel(fmt.Errorf("%v", r)) 
    5.   
  
  
  
  5.             } else { 
    6.   
  
  
  
  6.                 finish() 
    7.   
  
  
  
  7.             } 
    8.   
  
  
  
  8.         }() 
    9.   
  
  
  
  9.         reducer(collector, writer, cancel) 
    10.   
  
  
  
  10.         //这里 
    11.   
  
  
  
  11.         drain(collector) 
    12.   
  
  
  
  12.     }() 
    13.

这里的 reducer(collector, writer, cancel) 其实就是从 MapReduceVoid 传递的第三个闭包函数。

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

然后这个闭包函数又执行了 reducer(input, cancel)，这里的 reducer 就是我们一开始解释过的 VoidReducerFunc，从 Finish() 而来。

等等,看到上面三个地方的 drain(input)了吗?

 
 
 
 
    
  
  
  
  
  1. // drain drains the channel. 
    2.   
  
  
  
  2. func drain(channel <-chan interface{}) { 
    3.   
  
  
  
  3.     // drain the channel 
    4.   
  
  
  
  4.     for range channel { 
    5.   
  
  
  
  5.     } 
    6.   
  
  
  
  6. } 
    7.

其实就是一个排空 channel 的操作，但是三个地方都对同一个 channel做同样的操作，也是让我费解。

还有更重要的一点。

 
 
 
 
    
  
  
  
  
  1. go func() { 
    2.   
  
  
  
  2.         defer func() { 
    3.   
  
  
  
  3.             if r := recover(); r != nil { 
    4.   
  
  
  
  4.                 cancel(fmt.Errorf("%v", r)) 
    5.   
  
  
  
  5.             } else { 
    6.   
  
  
  
  6.                 finish() 
    7.   
  
  
  
  7.             } 
    8.   
  
  
  
  8.         }() 
    9.   
  
  
  
  9.         reducer(collector, writer, cancel) 
    10.   
  
  
  
  10.         drain(collector) 
    11.   
  
  
  
  11.     }() 
    12.

上面的代码，假如执行 reducer，writer 写入引发 panic,那么drain(collector) 将没有机会执行。

不过作者已经修复了这个问题，直接把 drain(collector) 放入到 defer。

具体 issues[1]。

到这里，关于 Finish 的源码也就结束了。感兴趣的可以看看其他源码。

很喜欢 go-zero 里的一些工具，但是工具往往并不独立，依赖于其他文件包，导致明明只想使用其中一个工具却需要安装整个包。

所以最终的结果就是扒源码，创建无依赖库工具集，遵循 MIT 即可。

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

 

本文题目：如何优雅地实现并发编排任务
URL标题：http://www.shufengxianlan.com/qtweb/news0/75100.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

动态网站知识

同城分类信息

成都二枢机房　　 　网络推广公司　　 　南充机房托管　　 　腾讯云免备案空间　　 　重庆网站设计　　 　建站公司　　 　域名ssl证书　　 　服务器租用　　 　成都能净科技　　 　全网整合营销推广　　 　宜宾网站建设　　 　成都展柜制作　　 　小程序开发公司　　 　重庆电信五里店机房　　 　网上商城管理系统　　 　一带一路　　 　成都搬家公司　　 　成都手机网站开发　　 　成都网站建设　　 　成都标识标牌设计制作