go-zero 成长之路—微服务电商实战系列（七、并发处理工具MapReduce的使用）

原创

GoLang成长之路

修改于 2022-10-20 18:13:04

5560

修改于 2022-10-20 18:13:04

文章被收录于专栏：go-zero微服务电商go-zero微服务电商

该系列源码已开源：
枫零落/microShop

1. 概述

大家都知道：go-zero的api服务需要通过api文件进行定义。其中返回值的类型定义中并没有支持time.Time的类型。

那问题来了：如果在返回值中需要时间的结果，针对sqlx和gorm模型定义的time.Time的时间类型应该怎么处理呢？

接下来咱们就以产品服务版块中的产品列表的返回结果来说明如何解决的！

如果历史文章不是很清楚的，可通过如下传送门：

2. `go-zero`返回结果中实现 `time.Time` 的处理

在这里咱们用到了：并发处理工具MapReduce，具体的工具介绍参考：https://legacy.go-zero.dev/cn/mapreduce.html

MapReduce 是 go-zero 框架中自带的，不需要额外安装。

MapReduce使用注意事项

mapper和reducer中都可以调用cancel，参数为error，调用后立即返回，返回结果为nil, error
mapper中如果不调用writer.Write则item最终不会被reducer聚合
reducer中如果不调用writer.Wirte则返回结果为nil, ErrReduceNoOutput
reducer为单线程，所有mapper出来的结果在这里串行聚合

使用流程：

generate 产生需要使用数据
mapper 处理 generate 产生的数据
mapper 对 mapper 的数据进行聚合处理，然后输出到结果
mapper 返回的结果：interface{} 通过 json.Marshal 、json.Unmarshal 处理成想要的数据

完整代码如下：

productsData, productsDataErr := mr.MapReduce(func(source chan<- interface{}) {
	// 这里是 generate | 将列表的下标值记录到 chan
	// 传入到 mapper
	for key := range products {
		source <- key
	}
}, func(item interface{}, writer mr.Writer, cancel func(error)) {
	// 这里是 mapper | 处理存入的 chan
	key := item.(int)
	// 写入到 reducer
	writer.Write(key)
}, func(pipe <-chan interface{}, writer mr.Writer, cancel func(error)) {
	// 处理 reducer | 对 mapper 进行数据聚合
	productsData := []types.ProductItem{}
	productsDataMap := []any{}
	for p := range pipe {
		itemOne := products[p.(int)]
		FormatString := "2006-01-02 15:04:05"
		product := types.ProductItem{
			ID:          int64(itemOne.Id),
			Title:       itemOne.Title,
			CoverPic:    itemOne.CoverPic,
			Property:    itemOne.Property,
			MtPrice:     itemOne.MtPrice,
			DisPrice:    itemOne.DisPrice,
			Stock:       int64(itemOne.Stock),
			State:       int64(itemOne.State),
			SalesVolume: itemOne.SalesVolume,
			Images:      itemOne.Images,
			Detail:      itemOne.Detail,
			CreateTime:  itemOne.CreateTime.Format(FormatString),
			UpdateTime:  itemOne.UpdateTime.Format(FormatString),
		}
		// 插入 []数组
		productsDataMap = append(productsDataMap, product)
		// interface 转  []types.ProductItem{}
		productsDataJson, _ := json.Marshal(productsDataMap)
		json.Unmarshal(productsDataJson, &productsData)
	}
	// 输出到结果
	writer.Write(productsData)
})

if productsDataErr != nil {
	return nil, errorx.NewDefaultError(productsDataErr.Error())
}

productsDataList := []*types.ProductItem{}
// 对 MapReduce 结果进行转换 | 因为 MapReduce 返回的结果是 interface{} 跟咱们返回的Data类型不一致
// interface 转  []*types.ProductItem{}
productsDataJson, _ := json.Marshal(productsData)
json.Unmarshal(productsDataJson, &productsDataList)