当前位置：首页 > 开源项目 > 微服务电商 > 正文内容

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

3年前 (2022-10-20)微服务电商15872

> 该系列源码已开源：
>
> [![枫零落/microShop](https://gitee.com/jobhandsome/micro-shop/widgets/widget_card.svg?colors=4183c4,ffffff,ffffff,e3e9ed,666666,9b9b9b)](https://gitee.com/jobhandsome/micro-shop)

# 1. 概述

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

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

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

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

- [go-zero 成长之路—微服务电商实战系列（六、条件查询）](https://juejin.cn/post/7155853768457158692)
- [go-zero 成长之路—微服务电商实战系列（五、RPC定义）](https://juejin.cn/post/7151936032508739615)
- [go-zero 成长之路—微服务电商实战系列（四、API定义）](https://juejin.cn/post/7142848286330880014)
- [go-zero 成长之路—微服务电商实战系列（三、表结构篇）](https://juejin.cn/post/7139717722782466084)
- [go-zero 成长之路—微服务电商实战系列（二、划分篇）](https://juejin.cn/post/7138610986612686856)
- [go-zero 成长之路—微服务电商实战系列（一、需求篇）](https://juejin.cn/post/7137280656912384036)

# 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`出来的结果在这里串行聚合

使用流程：

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

> 完整代码如下：

```go
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)
```

# 3. 结束语

本篇文章介绍说明了在开发过程中会出现的一些清情况：

*   检索条件的组合
*   分页条数的计算
*   检索条件的`sql`语句格式化输出
*   数据的返回

另外，如果你感兴趣，非常欢迎你加入，我们一起来完成这个项目，为社区献出自己的一份力。

希望本篇文章对你有所帮助，谢谢。