Background
In microservices development, the API gateway plays the role of providing restful API to the outside world, and the response data of the requests often relies on other services, and complex APIs even rely on multiple services or even dozens of services. Although the response time of a single dependent service is generally not slow, but if multiple services depended serially, then the response time of the API requests will be greatly increased.
So what are the best ways to optimize the response time? The first thing that comes to mind is to handle the dependencies in a concurrent manner, which reduces the time spent on the entire dependent services.
Although the go language performs great on lightweight concurrency, concurrent programming will always be one of the hardest problems in programming.
The Go standard library provides us with the sync.WaitGroup for concurrency control, but in actual scenarios, if one of the multiple dependencies is wrong, we expect to be able to return immediately rather than waiting for all the dependencies to be executed and then return the results, and the WaitGroup often requires locks on the variable assignments, and each related function needs to call Add and Done. For the beginners. It is error-prone.
Implementation Ideas
MapReduce is a software architecture proposed by Google for parallel computing on large-scale datasets, and the MapReduce tool in go-zero draws on this architectural idea.
The MapReduce tool in the go-zero is mainly used for concurrent processing of batch data to improve the performance of the service.
Use cases
Scenario 1
The result of some function often needs to rely on multiple services, for example, the result of the product detail often depends on the user service, inventory service, order service, etc. Generally, the dependent services are provided as rpc services, in order to reduce the response time, we often need to call the dependencies in parallel.
func productDetail(uid, pid int64) (*ProductDetail, error) {
var pd ProductDetail
err := mr.Finish(func() error {
u, err := getUserInfo(uid)
if err ! = nil {
return err
}
pd.User = u
pd.User = u
}, func() error {
if s, err := getStoreInfo(pid) if err != nil {
return err
}
pd.Store = s
return nil
})
if err ! = nil {
return nil, err
}
return &pd, nil
}Scenario 2
Many times we need to process a batch of data, such as a batch of user ids, check the legitimacy of each user and if there is an error in the checking process, it is considered that the checking has failed, and the returned result is the user ids that are legitimized.
func checkLegal(uids []int64) ([]int64, error) {
v, err := mr.MapReduce(func(source chan<- int64) {
for _, uid := range uids {
source <- uid
}
}, func(v int64, writer mr.Writer[int64], cancel func(error)) {
err := checkUser(v)
if err != nil {
//cancel(err) // 会阻断
return
}
writer.Write(v)
}, func(pipe <-chan int64, writer mr.Writer[[]int64], cancel func(error)) {
var validUids []int64
for v := range pipe {
validUids = append(validUids, v)
}
writer.Write(validUids)
})
return v, err
}func checkLegal2(uids []int64) ([]int64, error) {
var validUids []int64
err := mr.MapReduceVoid(func(source chan<- int64) {
for _, uid := range uids {
source <- uid
}
}, func(v int64, writer mr.Writer[int64], cancel func(error)) {
err := checkUser(v)
if err != nil {
//cancel(err)
return
}
writer.Write(v)
}, func(pipe <-chan int64, cancel func(error)) {
for uid := range pipe {
validUids = append(validUids, uid)
}
}, mr.WithWorkers(20))
return validUids, err
}Scenario 3
Many times we need to process a batch of data, such as a batch of user ids, to update them.
func updateUsers(uids []int64) {
mr.ForEach(func(source chan<- int64) {
for _, uid := range uids {
source <- uid
}
}, func(item int64) {
if err := updateUser(item); err != nil {
log.Println(err)
}
}, mr.WithWorkers(20))
}Important notes
- mr.Finish
- Finish will block if there is an error in the process
- Will not recover panics
2. mr.FinishVoid
- FinishVoid will not be blocked if there is an error in the process.
- Will not recover panics
3. mr.MapReduce, MapReduceVoid, MapReduceChan
- MapReduce, MapReduceVoid, MapReduceChan, need to determine whether to block, need to manually call cancel(err)
- Will not recover panics
- Default concurrency 16, if you need to adjust concurrency, you can set it via mr.WithWorkers, but it must be > 1 to be useful.
4. mr.ForEach
- ForEach will not block if there is an error in the process
- Will not recover panics
- Default concurrency 16, if you need to adjust the concurrency, you can set it via mr.WithWorkers, but it must be > 1 to be useful.
Project address
https://github.com/zeromicro/go-zero
Feel free to use go-zero and star to support us!
