go zero 使用MapReduce并发
一、MapReduce 介绍
MapReduce 是一种用于并行计算的编程模型，特别适合在大规模数据处理场景中简化逻辑代码。
官方文档：
https://go-zero.dev/docs/components/mr
1. MapReduce 的核心概念
在 MapReduce 中，主要有以下三个核心步骤：
a. Generate (生成数据)：
- 数据的初始输入阶段。可以是一个简单的循环，也可以是从数据库、文件或其他来源加载数据。
b. Mapper (映射)： - 将输入数据映射为中间结果。通常用来过滤、转换、查询或处理数据。
c. Reducer (归约)： - 对映射后的数据进行汇总处理，生成最终的结果。
在 go zero 中，mr.MapReduce 的具体代码如下：
func MapReduce[T, U, V any](generate GenerateFunc[T], mapper MapperFunc[T, U], reducer ReducerFunc[U, V],opts ...Option) (V, error) {panicChan := &onceChan{channel: make(chan any)}source := buildSource(generate, panicChan)return mapReduceWithPanicChan(source, panicChan, mapper, reducer, opts...)
}2. 为什么需要 MapReduce
在实际的业务场景中我们常常需要从不同的 rpc 服务中获取相应属性来组装成复杂对象。
比如要查询商品详情：
- 商品服务-查询商品属性
- 库存服务-查询库存属性
- 价格服务-查询价格属性
- 营销服务-查询营销属性
如果是串行调用的话响应时间会随着 rpc 调用次数呈线性增长，所以我们要优化性能一般会将串行改并行。
简单的场景下使用 WaitGroup 也能够满足需求，但是如果我们需要对 rpc 调用返回的数据进行校验、数据加工转换、数据汇总呢？继续使用 WaitGroup 就有点力不从心了.。
二、项目构建
接下来我们使用一个文章列表功能简单的演示下
1. article数据表
这是存储文章信息的表，包含标题、内容、作者、评论数等字段。
CREATE TABLE `article` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键ID',`title` VARCHAR(255) NOT NULL DEFAULT '' COMMENT '标题' COLLATE 'utf8mb4_bin',`content` TEXT NOT NULL COMMENT '内容' COLLATE 'utf8_unicode_ci',`cover` VARCHAR(255) NOT NULL DEFAULT '' COMMENT '封面' COLLATE 'utf8mb4_bin',`description` VARCHAR(255) NOT NULL DEFAULT '' COMMENT '描述' COLLATE 'utf8mb4_bin',PRIMARY KEY (`id`)
);
2.article.api
在实际开发中，应该传入作者ID、游标、页码、排序方法等信息，这里为了方便演示就不传入参数了，API 定义如下：
syntax = "v1"type (ArticleInfo {ArticleId int64 `json:"article_id"`Title string `json:"title"`Content string `json:"content"`Description string `json:"description"`Cover string `json:"cover"`}ArticleListResponse {Articles []ArticleInfo `json:"articles"`}
)@server (prefix: /v1/article
)
service article-api {@handler Articlelisthandlerpost /list returns (ArticleListResponse)
}
三、使用 MapReduce
拉取库
go get github.com/zeromicro/go-zero/core/mr
1.实现文章列表
我们使用 MapReduce 来并行处理文章数据，
func (l *ArticlelistLogic) Articlelist() (resp *types.ArticleListResponse, err error) {// Step 1: Generate 数据//这里为了方便我使用了简单for循环产生文档IDgenerateFunc := func(source chan<- int) {for id := 1; id < 50; id++ { // 模拟文章 ID 数据source <- id}}articleModel := l.svcCtx.ArticleModel// Step 2: Mapper 映射处理mapperFunc := func(id int, writer mr.Writer[*types.ArticleInfo], cancel func(error)) {//使用产生id，查询文章详情one, err := articleModel.FindOne(l.ctx, uint64(id)) // 查找单篇文章if err != nil {return // 跳过错误}//FindOne返回的是 *model.Article类型，Mapper映射的类型为*types.ArticleInfo//所以需要转换一下articleInfo := &types.ArticleInfo{ArticleId: int64(one.Id),Title: one.Title,Content: one.Content,Description: one.Description,Cover: one.Cover,}writer.Write(articleInfo) // 写入中间结果}// Step 3: Reducer 汇总处理reduceFunc := func(pipe <-chan *types.ArticleInfo, writer mr.Writer[[]types.ArticleInfo], cancel func(error)) {var articleList []types.ArticleInfofor article := range pipe {articleList = append(articleList, *article) }writer.Write(articleList) // 写入最终结果}// 调用 MapReduce//mr.WithWorkers(5) 允许调用者自定义并发工作线程数。//如果不传入mr.WithWorkers ，默认Workers为16个reduce, err := mr.MapReduce(generateFunc, mapperFunc, reduceFunc, mr.WithWorkers(5)) if err != nil {return nil, err // 处理错误}// 返回结果return &types.ArticleListResponse{Articles: reduce,}, nil
}
2. 详细讲解
Step 1: Generate 数据
generateFunc 的作用是提供初始数据。在本例中，我们通过一个循环生成了文章的 ID：
generateFunc := func(source chan<- int) {for id := 1; id < 50; id++ {source <- id}
}
Step 2: Mapper 映射处理
mapperFunc 用于处理每一个文章 ID，并将其转换为 ArticleInfo。
- 使用
articleModel.FindOne从数据库中获取文章数据。 - 如果获取失败，跳过该 ID。
- 将结果通过
writer.Write写入到下一步。
mapperFunc := func(id int, writer mr.Writer[*types.ArticleInfo], cancel func(error)) {one, err := articleModel.FindOne(l.ctx, uint64(id))if err != nil {return}articleInfo := &types.ArticleInfo{ArticleId: int64(one.Id),Title: one.Title,Content: one.Content,Description: one.Description,Cover: one.Cover,}writer.Write(articleInfo)
}
Step 3: Reducer 汇总处理
reduceFunc 将 mapperFunc 的结果汇总为最终的 []types.ArticleInfo。
- 遍历管道中的每个
*types.ArticleInfo。 - 将解引用后的
ArticleInfo添加到结果列表。
reduceFunc := func(pipe <-chan *types.ArticleInfo, writer mr.Writer[[]types.ArticleInfo], cancel func(error)) {var articleList []types.ArticleInfofor article := range pipe {articleList = append(articleList, *article)}writer.Write(articleList)
}
3. 测试运行
向 /v1/article/list 发送 POST 请求：
curl -X POST http://localhost:8888/v1/article/list
运行结果如下：
{"articles": [{"article_id": 1,"title": "标题1","content": "这是内容1","description": "描述1","cover": "封面1.jpg"},...]
}
4.效率对比
普通循环
为了更直观的对比效率，我们使用普通循环再次实现下文章列表：
func (l *ArticlelistLogic) Articlelist() (resp *types.ArticleListResponse, err error) {// todo: add your logic here and delete this linetime1 := time.Now()var articleList []types.ArticleInfoarticleModel := l.svcCtx.ArticleModelfor id := 1; id < 50; id++ {article, _ := articleModel.FindOne(l.ctx, uint64(id))articleInfo := types.ArticleInfo{ArticleId: int64(article.Id),Title: article.Title,Content: article.Content,Description: article.Description,Cover: article.Cover,}articleList = append(articleList, articleInfo)}time2 := time.Now()logx.Info("执行时间为：", time2.Sub(time1))return &types.ArticleListResponse{Articles: articleList,}, nil}效率对比
这个执行时间可能每次都不一样，但是进过多次对比， 使用mapreduce 效率是高于普通方法的
使用串行调用时间：
使用MapReduce消耗时间：
