搜索

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

Go语言精选

共 17690字，需浏览 36分钟

 · 2021-05-15

业务场景


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


场景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


推荐阅读


福利

我为大家整理了一份从入门到进阶的Go学习资料礼包，包含学习建议：入门看什么，进阶看什么。关注公众号 「polarisxu」，回复 ebook 获取；还可以回复「进群」，和数万 Gopher 交流学习。

浏览 32
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
偷偷告诉你如何一台电脑开多个微信！
大家好，我是轩辕。前几天在粉丝群里，有人问我是怎么在一台电脑上同时登录两个微信的？正好之前写过一篇文章，分析过原理，分享给没看过的小伙伴学习一下。手机端多开微信估计很多人都知道，像华为、小米等手机系统都对此做了支持，不过在运行Windows系统的电脑上怎么启动两个微信呢？其实很简单，你只需要写一个批
编程技术宇宙
0
Go Heap Profile 怎么了？
Go heap profile 是常常用来检查内存泄露和内存占用大问题的问题的手段，而且非常常用。而且，我们也经常创建两个间隔较长的 heap profile, 获取它们的差值来方便查看内存泄露: Hi, 使用多年的 go pprof 检查内存泄漏的方法居然是错的?! [1]今天，度厂的一位同学提出
GoCN
0
阿里的同事，写的代码真 TMD 优雅！
通过这篇文章你将了解到整洁的代码对项目、公司和你的重要性,以及如何书写整洁的代码.通过命名、类、函数、测试这四个章节,使我们的代码变得整洁.1、为什么要保持代码整洁?不整洁的代码随着时间的增加而增加时,生产力会随之降低.导致的结果就是:代码不易扩展或扩展容易引发其他问题程序崩溃加班增加公司成本(加人
Java专栏
1
测试新人，如何快速上手一个陌生的系统！
大家好，我是狂师！作为刚入行不久的测试新人，面对一个陌生的系统时，可能会感到有些手足无措。面对一个全新的系统系统，如何快速上手并展开有效的测试工作是一个重要的挑战。本文将探讨测试新人如何通过一系列步骤和策略，快速熟悉并掌握新系统的测试要点，从而提高测试效率和质量。本文旨在为测试新手提供一份指导，帮助
测试开发技术
0
光纤详解：光纤跳线如何分类，多向单模转换？
本文来自“光纤详解：光纤跳线如何分类，多向单模转换？”，光纤跳线作为光网络布线最基础的元件之一，被广泛应用于光纤链路的搭建中。如今，光纤制造商根据应用场景的不同推出众多类型的光纤跳线，如MPO/LC/SC/FC/ST光纤跳线，单工/双工光纤跳线，单模/多模光纤跳线等，它们之间各有特色，且不可替代。本
架构师技术联盟
0
如何计算数据中心的冷却需求？
  今日分享  【导读】数据中心的冷却要求受多种因素影响，包括设备的热量输出、占地面积、设施设计和电气系统功率额定值等等……众所周知，环境因素会严重影响数据中心设备。过多的热量积聚会损坏服务器，可能导致其自动关闭。经常在高于可接受的温度下运行服务器会缩短其使用
数据中心运维管理
0
5000w+ 的大表如何拆？亿级别大表拆分实战复盘
前言笔者是在两年前接手公司的财务系统的开发和维护工作。在系统移交的初期，笔者和团队就发现，系统内有一张5000W+的大表。跟踪代码发现，该表是用于存储资金流水的表格，关联着众多功能点，同时也有众多的下游系统在使用这张表的数据。进一步的观察发现，这张表还在以每月600W+的数据持续增长，也就是说，不超
码农编程进阶笔记
0
多人同时导出 Excel 干崩服务器！新来的阿里大佬给出的解决方案太优雅了！
点击关注公众号，Java 干货及时推送↓推荐阅读：面试辅导，我们出大成果了！来源：juejin.cn/post/7259249904777838629前言 业务诉求：考虑到数据库数据日渐增多，导出会有全量数据的导出，多人同时导出可以会对服务性能造成影响，导出涉及到mysql查询的io操作，
Java技术栈
1
如何做到无感刷新Token?
来源：juejin.cn/post/7316797749517631515为什么需要无感刷新Token？自动刷新token前端token续约疑问及思考图片为什么需要无感刷新Token？「最近浏览到一个文章里面的提问，是这样的：」当我在系统页面上做业务操作的时候会出现突然闪退的情况，然后跳转到登录页面
Java专栏
2
15种时间序列预测方法总结(包含多种方法代码实现)
向AI转型的程序员都关注了这个号👇👇👇在这篇文章中，我们将深入探讨时间序列预测的基本概念和方法。我们将首先介绍单元预测和多元预测的概念，然后详细介绍各种深度学习和传统机器学习方法如何应用于时间序列预测，包括循环神经网络（RNN）、一维卷积神经网络（1D-CNN）、Transformer、自回归模型（
机器学习AI算法工程
0
点赞
评论
收藏
分享

手机扫一扫分享

举报