文章持续更新,微信搜一搜「 吴亲强的深夜食堂 」
业务场景
在做任务开发的时候,你们一定会碰到以下场景:
场景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
}
Gsync.WaitGroup
好像,问题不大。
996
那么我们能不能抽像出一套针对此业务场景的工具,而把具体业务实现交给业务方。
安排。
使用
go-zeromapreduceGoogle
使用很简单。我们通过它改造一下上面的代码:
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
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
GenerateFuncMapperFuncVoidReducerFuncmapper
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 //返回无缓冲通道
}
buildSourceGgenerate(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
}
}
GexecuteMappers
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)
})
}
}
}
具体的逻辑已备注,代码很容易懂。
executeMapperscollectorreducer
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)reducerVoidReducerFuncFinish() 而来
drain(input)
// drain drains the channel.
func drain(channel <-chan interface{}) {
// drain the channel
for range channel {
}
}
channelchannel
还有更重要的一点。
go func() {
defer func() {
if r := recover(); r != nil {
cancel(fmt.Errorf("%v", r))
} else {
finish()
}
}()
reducer(collector, writer, cancel)
drain(collector)
}()
reducerwriterpanicdrain(collector)
drain(collector)defer
具体 issues[1]。
Finish
go-zero