Go-Zero Concurrency Tool: Detailed Analysis of `mr` Package Functions and Task Concurrency Execution Principles

The go-zero/core/mr package provides a set of tools designed to simplify concurrent programming in Go, particularly for the MapReduce pattern and lightweight task synchronization control.

I. Core MapReduce Model

MapReduce is a classic data processing model used for processing and aggregating large-scale data in parallel environments.

1. `mr.MapReduce`

Attribute	Description	Source
Purpose/Principle	The classic MapReduce model. Used for complex parallel computing, data transformation, and aggregation. It processes data flow through three stages.
Structure	Accepts three main functions as parameters: `generate`, `mapper`, and `reducer`.
Return	Returns the final aggregated result (`result`) and error (`err`).

Implementation Principle (Three Stages):

generate: Generate Data Source. Responsible for feeding raw data into a channel (source chan<- interface{}), consumed by the Mapper.
mapper: Concurrent Processing. Concurrently processes each element (item) provided by generate and outputs one or more intermediate results via the writer.
reducer: Aggregate Results. Receives all Mapper outputs from the pipeline (pipe <-chan interface{}), performs final aggregation, and returns the final result via the writer.

How to Use (Example: Calculating Sum of Squares):

result, err := mr.MapReduce(
// 1. generate: Generates 1 to 5
func(source chan<- interface{}) {
for i := 1; i <= 5; i++ {
source <- i
}
},
// 2. mapper: Calculates the square of the element
func(item interface{}, writer mr.Writer[interface{}], cancel func(error)) {
num := item.(int)
writer.Write(num * num) // Output the square result
},
// 3. reducer: Aggregates the sum
func(pipe <-chan interface{}, writer mr.Writer[interface{}], cancel func(error)) {
var sum int
for item := range pipe {
sum += item.(int)
}
writer.Write(sum) // Return the final sum
},
)
// The final result is 55

II. MapReduce Variants

1. `mr.MapReduceChan`

Attribute	Description	Source
Purpose/Principle	Similar to `mr.MapReduce`, but its data source is an existing channel instead of being generated by a `generate` function.
Structure	Accepts a channel as the data source, along with the `mapper` and `reducer` functions.

How to Use (Example: Reading from a pre-filled channel):

source := make(chan interface{}, 5)
// Fill and close source channel
// ...
result, err := mr.MapReduceChan(
source, // Directly pass the channel
// ... mapper and reducer logic is the same as mr.MapReduce
)

2. `mr.MapReduceVoid`

Attribute	Description	Source
Purpose/Principle	Similar to `mr.MapReduce`, but it does not return a final result. Typically used in scenarios where only the `mapper` operation needs to be executed (e.g., side effects, writing to a database).

III. Concurrent Iteration and Control

1. `mr.ForEach`

Attribute	Description	Source
Purpose/Principle	Focuses on data iteration. It concurrently executes the `mapper` function for each element generated by the `generate` function. Primarily focused on the concurrency of data processing.
Structure	Accepts the `generate` function (to produce the data source) and the `mapper` function (to process a single element).
Return	No return value (`void`).

How to Use (Example: Concurrent processing of 5 elements):

var count int32
mr.ForEach(
// generate: Generates elements 0 to 4
func(source chan<- interface{}) {
for i := 0; i < 5; i++ {
source <- i
}
},
// mapper: Concurrently processes each element
func(item interface{}) {
atomic.AddInt32(&count, 1)
fmt.Printf("Processing element: %d\n", item.(int))
},
)
// The final count should equal 5

IV. Task Synchronization and Error Control

These functions are used to concurrently execute a group of independent tasks and provide fine-grained control over error propagation.

1. `mr.Finish`

Attribute	Description	Source
Purpose/Principle	Concurrently executes a set of independent tasks, emphasizing Fast Fail (fail-fast) and error/panic capture.
Fast Fail Principle	If any one of the functions returns an error, it immediately cancels all other currently executing functions and returns that error.
Panic Handling Principle	Capable of capturing a `panic` that occurs during task execution, and encapsulates it into an error. The Go-Zero library prepends "panic: " to the error message.
Successful Return	If all functions execute successfully, it returns `nil`.

How to Use (Example: Task failure leading to quick exit):

err := mr.Finish(
func() error { fmt.Println("Task 1 complete"); return nil },
func() error {
time.Sleep(10 * time.Millisecond);
fmt.Println("Task 2 failed");
return errors.New("Task 2 failed")
},
func() error { // This task waits a long time but will be cancelled
time.Sleep(100 * time.Millisecond);
fmt.Println("Task 3 cancelled");
return nil
},
)
// err is not nil, and equals "Task 2 failed"

How to Use (Example: Panic capture):

err := mr.Finish(
// ... Other tasks ...
func() error {
panic("Task 3 encountered a panic")
},
)
// The captured error message is "panic: Task 3 encountered a panic"

2. `mr.FinishVoid`

Attribute	Description	Source
Purpose/Principle	Similar to `mr.Finish`, but it does not return any error. Suitable for scenarios where tasks need to be executed concurrently without the need for error handling or fast failure.
Synchronization Principle	It waits for all the functions passed in to complete their execution.

How to Use (Example: Concurrent execution of tasks with no return values):

var count int32
mr.FinishVoid(
func() { atomic.AddInt32(&count, 1) },
func() { atomic.AddInt32(&count, 1) },
func() { atomic.AddInt32(&count, 1) },
)
// The final count should equal 3

V. Configuration Options (Options)

These functions can be passed as optional parameters to mr.MapReduce or mr.ForEach to control the behavior of the concurrent process.

1. `mr.WithContext(ctx)`

Attribute	Description	Source
Purpose/Principle	Allows a `context.Context` to be passed, thus enabling control over the MapReduce process’s lifecycle. This is very useful for implementing timeout or external cancellation mechanisms.

How to Use (Example: Setting a timeout):

ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()

_, err := mr.MapReduce(
// ... generate / mapper / reducer definitions ...
mr.WithContext(ctx), // Pass the context with timeout
)
// If the operation exceeds 50ms, err will be context.DeadlineExceeded

2. `mr.WithWorkers(n)`

Attribute	Description	Source
Purpose/Principle	Allows the user to specify the number of goroutines (concurrent workers) that execute the `mapper`. If not specified, the number of CPU cores is used by default.

How to Use (Example: Specifying 4 workers):

mr.ForEach(
// ... generate 10 elements ...
// ... mapper logic ...
mr.WithWorkers(4), // Specify 4 concurrent workers
)

技术博客 | Go & Kubernetes & Linux

I. Core MapReduce Model ​

1. mr.MapReduce ​

II. MapReduce Variants ​

1. mr.MapReduceChan ​

2. mr.MapReduceVoid ​

III. Concurrent Iteration and Control ​

1. mr.ForEach ​

IV. Task Synchronization and Error Control ​

1. mr.Finish ​

2. mr.FinishVoid ​

V. Configuration Options (Options) ​

1. mr.WithContext(ctx) ​

2. mr.WithWorkers(n) ​