go-zero/core/mr 包提供了一系列用于简化 Go 语言中并发编程的工具，特别是针对 MapReduce 模式和轻量级任务同步控制。
I. 核心 MapReduce 模型
MapReduce 是一种经典的数据处理模型，用于在并行环境中处理和聚合大规模数据。
1. mr.MapReduce
| 属性 | 描述 | 来源 |
|---|---|---|
| 用途/原理 | 经典的 MapReduce 模型。用于复杂的并行计算、数据转换和聚合。它通过三个阶段实现数据流的处理。 | |
| 结构 | 接受三个主要函数作为参数：generate、mapper 和 reducer。 | |
| 返回 | 返回最终聚合的结果 (result) 和错误 (err)。 |
实现原理 (三阶段):
generate: 生成数据源。负责将原始数据送入一个 channel (source chan<- interface{})，供 Mapper 消费。mapper: 并发处理。并发地处理generate提供的每个元素 (item)，并通过writer输出一个或多个中间结果。reducer: 聚合结果。从管道 (pipe <-chan interface{}) 中接收所有 Mapper 的输出，进行最终聚合，并通过writer返回最终结果。
如何使用 (示例：计算平方和):
go
result, err := mr.MapReduce(
// 1. generate: 生成 1 到 5
func(source chan<- interface{}) {
for i := 1; i <= 5; i++ {
source <- i
}
},
// 2. mapper: 计算元素的平方
func(item interface{}, writer mr.Writer[interface{}], cancel func(error)) {
num := item.(int)
writer.Write(num * num) // 输出平方结果
},
// 3. reducer: 聚合总和
func(pipe <-chan interface{}, writer mr.Writer[interface{}], cancel func(error)) {
var sum int
for item := range pipe {
sum += item.(int)
}
writer.Write(sum) // 返回最终总和
},
)
// 最终 result 为 55II. MapReduce 变体
1. mr.MapReduceChan
| 属性 | 描述 | 来源 |
|---|---|---|
| 用途/原理 | 与 mr.MapReduce 类似，但其数据源不是通过 generate 函数生成，而是一个已经存在的 channel。 | |
| 结构 | 接受一个 channel 作为数据源，以及 mapper 和 reducer 函数。 |
如何使用 (示例：从预先填充的 channel 读取):
go
source := make(chan interface{}, 5)
// 填充并关闭 source channel
// ...
result, err := mr.MapReduceChan(
source, // 直接传入 channel
// ... mapper 和 reducer 逻辑与 mr.MapReduce 相同
)2. mr.MapReduceVoid
| 属性 | 描述 | 来源 |
|---|---|---|
| 用途/原理 | 与 mr.MapReduce 类似，但它不返回最终结果。通常用于只需要执行 mapper 操作（如副作用操作、写入数据库等）的场景。 |
III. 并发迭代与控制
1. mr.ForEach
| 属性 | 描述 | 来源 |
|---|---|---|
| 用途/原理 | 专注于数据迭代。它会为 generate 函数生成的每个元素并发执行 mapper 函数。主要关注数据处理的并发性。 | |
| 结构 | 接受 generate 函数（生成数据源）和 mapper 函数（处理单个元素）。 | |
| 返回 | 无返回值（void）。 |
如何使用 (示例：并发处理 5 个元素):
go
var count int32
mr.ForEach(
// generate: 生成 0 到 4 的元素
func(source chan<- interface{}) {
for i := 0; i < 5; i++ {
source <- i
}
},
// mapper: 并发处理每个元素
func(item interface{}) {
atomic.AddInt32(&count, 1)
fmt.Printf("处理元素: %d\n", item.(int))
},
)
// 最终 count 应该等于 5IV. 任务同步与错误控制
这些函数用于并发执行一组独立任务，并对错误传播进行精细控制。
1. mr.Finish
| 属性 | 描述 | 来源 |
|---|---|---|
| 用途/原理 | 并发执行一组独立的任务，强调快速失败 (fail-fast) 和错误/恐慌捕获。 | |
| 快速失败原理 | 如果其中任何一个函数返回错误，它会立即取消所有其他正在执行的函数并返回该错误。 | |
| 恐慌处理原理 | 能够捕获任务执行期间发生的 panic（恐慌），并将其封装成一个错误返回。Go-Zero 库会在错误信息前加上 "panic: " 前缀。 | |
| 成功返回 | 如果所有函数都成功执行，返回 nil。 |
如何使用 (示例：任务失败导致快速退出):
go
err := mr.Finish(
func() error { fmt.Println("任务1完成"); return nil },
func() error {
time.Sleep(10 * time.Millisecond);
fmt.Println("任务2失败");
return errors.New("任务2发生错误")
},
func() error { // 该任务等待时间长，但会被取消
time.Sleep(100 * time.Millisecond);
fmt.Println("任务3被取消");
return nil
},
)
// err 不为 nil，并且等于 "任务2发生错误"如何使用 (示例：恐慌捕获):
go
err := mr.Finish(
// ... 其他任务 ...
func() error {
panic("任务3发生恐慌")
},
)
// 捕获到的错误信息是 "panic: 任务3发生恐慌"2. mr.FinishVoid
| 属性 | 描述 | 来源 |
|---|---|---|
| 用途/原理 | 与 mr.Finish 类似，但它不返回任何错误。适用于只需要并发执行一组任务，而无需处理错误或快速失败的场景。 | |
| 同步原理 | 它会等待所有传入的函数执行完毕。 |
如何使用 (示例：并发执行非返回值的任务):
go
var count int32
mr.FinishVoid(
func() { atomic.AddInt32(&count, 1) },
func() { atomic.AddInt32(&count, 1) },
func() { atomic.AddInt32(&count, 1) },
)
// 最终 count 应该等于 3V. 配置选项 (Options)
这些函数可作为可选参数传递给 mr.MapReduce 或 mr.ForEach，用于控制并发流程的行为。
1. mr.WithContext(ctx)
| 属性 | 描述 | 来源 |
|---|---|---|
| 用途/原理 | 允许传入一个 context.Context，从而可以控制 MapReduce 流程的生命周期。这对于实现超时或外部取消机制非常有用。 |
如何使用 (示例：设置超时):
go
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
_, err := mr.MapReduce(
// ... generate / mapper / reducer 定义 ...
mr.WithContext(ctx), // 传入带超时的 context
)
// 如果操作超过 50ms，err 将是 context.DeadlineExceeded2. mr.WithWorkers(n)
| 属性 | 描述 | 来源 |
|---|---|---|
| 用途/原理 | 允许用户指定执行 mapper 的 goroutine 数量（并发工作者数量）。如果未指定，默认情况下会使用 CPU 核心数。 |
如何使用 (示例：指定 4 个 worker):
go
mr.ForEach(
// ... generate 10 个元素 ...
// ... mapper 逻辑 ...
mr.WithWorkers(4), // 指定 4 个并发 worker
)