MapReduce 和 Go 相关

One Worker Nine Watchers

找封面的时候突然看到一核有难多核围观，这下不得不改一张了

MapReduce 通过提供 MR 框架，为用户提供分布式计算的能力，用户需要为任务设计 Map 和 Reduce 函数。

MapReduce 对任务有要求，得是无状态的函数。

例如下方的 wc.go：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// The map function is called once for each file of input. The first
// argument is the name of the input file, and the second is the
// file's complete contents. You should ignore the input file name,
// and look only at the contents argument. The return value is a slice
// of key/value pairs.
func Map(filename string, contents string) []mr.KeyValue {
	// function to detect word separators.
	ff := func(r rune) bool { return !unicode.IsLetter(r) }

	// split contents into an array of words.
	words := strings.FieldsFunc(contents, ff)

	kva := []mr.KeyValue{}
	for _, w := range words {
		kv := mr.KeyValue{w, "1"}
		kva = append(kva, kv)
	}
	return kva
}

// The reduce function is called once for each key generated by the
// map tasks, with a list of all the values created for that key by
// any map task.
func Reduce(key string, values []string) string {
	// return the number of occurrences of this word.
	return strconv.Itoa(len(values))
}

目的是为了统计词频，输入是一批文件，输出是每个单词出现的次数。

Map 函数处理单个文件并返回一个 KeyValue 数组，Key 是单词，Value 是出现次数（这里是 1），Reduce 函数处理同一个 Key 的所有 Value 并返回总数。

注意到这里的 mr.KeyValue{w, "1"} 始终是 1，没在 Map 内部做累加。事实上也可以改为在 Map 内预先做累加，然后再在 Reduce 进一步累加各个 Map 的中间结果，得出最终结果。

那么被隐藏的部分就由 MapReduce 框架来完成了。MapReduce 有两个部分：Coordinator 和 Worker，Coordinator 负责分配任务，Worker 负责执行任务。

那么显然，由框架完成的部分包括分配任务、收集中间结果、分发 Reduce 任务、收集最终结果。

Map 输出的是 []mr.KeyValue，Reduce 输入的是 {string, []string} 显然中间还有步骤。

Map 的输出会被直接保存到文件中，例如：mr-<MapTaskID>-<ReduceTaskID>.json，Map 函数要预先知道哪些 Reduce 会分配哪些 Key 的处理，然后提前按照 Reduce 所属的 Key 分组写入不同的中间文件。（通常通过一个 Hash 函数来决定 Key 属于哪个 Reduce）

所有 Map 任务完成后，分配到 Reduce 任务的 Worker 要读取所有 Map 输出的中间文件中，与其 ReduceTaskID 匹配的中间文件，例如 mr-*-<ReduceTaskID>.json，归并排序后划分相同 Key 的全部 Value 到数组中，然后传入 Reduce 函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// mr-<MapTaskID>-<ReduceTaskID>.json

{"Key":"The","Value":"1"}
{"Key":"Project","Value":"1"}
{"Key":"Gutenberg","Value":"1"}
{"Key":"of","Value":"1"}
{"Key":"The","Value":"1"}
{"Key":"of","Value":"1"}
{"Key":"of","Value":"1"}
{"Key":"at","Value":"1"}
{"Key":"and","Value":"1"}
{"Key":"You","Value":"1"}
{"Key":"copy","Value":"1"}
{"Key":"terms","Value":"1"}
{"Key":"of","Value":"1"}
...

最后得到的结果是一串 Reduce 输出文件，例如 mr-out-<ReduceTaskID>。由于相同 Key 只会分配到同一个 Reduce 任务，因此最终输出的文件中不会有重复 Key。

MapReduce 框架容忍 Worker 故障，但是不特殊处理的话，Coordinator 故障就会导致整个任务失败，这是单点故障。

后来添加了主备机制

Coordinator 作为服务端接受来自 Worker 的请求，Worker 主动请求任务，汇报执行结果。Coordinator 对 Worker 有超时机制，超时则视为故障，将任务重新分配给其他空闲 Worker，非常简单。

以下是 make mr 的测试输出：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
go build -race -o main/mrsequential main/mrsequential.go
go build -race -o main/mrcoordinator main/mrcoordinator.go
go build -race -o main/mrworker main/mrworker.go&
(cd mrapps && go build -race -buildmode=plugin wc.go) || exit 1
(cd mrapps && go build -race -buildmode=plugin indexer.go) || exit 1
(cd mrapps && go build -race -buildmode=plugin mtiming.go) || exit 1
(cd mrapps && go build -race -buildmode=plugin rtiming.go) || exit 1
(cd mrapps && go build -race -buildmode=plugin jobcount.go) || exit 1
(cd mrapps && go build -race -buildmode=plugin early_exit.go) || exit 1
(cd mrapps && go build -race -buildmode=plugin crash.go) || exit 1
(cd mrapps && go build -race -buildmode=plugin nocrash.go) || exit 1
cd mr; go test -v -race 
=== RUN   TestWc
Coordinator state changed to ReducingState
Coordinator state changed to DoneState
--- PASS: TestWc (21.81s)
=== RUN   TestIndexer
2026/09/24 22:06:21 dialing:dial unix /tmp/5840-mr-Qnqbh_02ium0BKw1jRKs: connect: no such file or directory
2026/09/24 22:06:21 dialing:dial unix /tmp/5840-mr-Qnqbh_02ium0BKw1jRKs: connect: no such file or directory
Coordinator state changed to ReducingState
Coordinator state changed to DoneState
--- PASS: TestIndexer (8.30s)
=== RUN   TestMapParallel
2026/09/24 22:06:29 dialing:dial unix /tmp/5840-mr-YnVg0RKkVxWv1YG6Z2Cv: connect: no such file or directory
Coordinator state changed to ReducingState
Coordinator state changed to DoneState
--- PASS: TestMapParallel (8.06s)
=== RUN   TestReduceParallel
2026/09/24 22:06:37 dialing:dial unix /tmp/5840-mr-W5i9fmQlLCPCDA9dgS_Q: connect: no such file or directory
Coordinator state changed to ReducingState
Coordinator state changed to DoneState
--- PASS: TestReduceParallel (9.14s)
=== RUN   TestJobCount
2026/09/24 22:06:47 dialing:dial unix /tmp/5840-mr-S_xKyY1Z7I3xIZ67jDWd: connect: no such file or directory
Coordinator state changed to ReducingState
Coordinator state changed to DoneState
--- PASS: TestJobCount (12.04s)
=== RUN   TestEarlyExit
2026/09/24 22:06:59 dialing:dial unix /tmp/5840-mr-9wk5pOrYuj3kFMExsq5z: connect: no such file or directory
2026/09/24 22:06:59 dialing:dial unix /tmp/5840-mr-9wk5pOrYuj3kFMExsq5z: connect: no such file or directory
2026/09/24 22:06:59 dialing:dial unix /tmp/5840-mr-9wk5pOrYuj3kFMExsq5z: connect: no such file or directory
Coordinator state changed to ReducingState
Coordinator state changed to DoneState
2026/09/24 22:07:06 dialing:dial unix /tmp/5840-mr-e8lKxZ_E4zdfQJZYtZ5J: connect: no such file or directory
2026/09/24 22:07:06 dialing:dial unix /tmp/5840-mr-e8lKxZ_E4zdfQJZYtZ5J: connect: no such file or directory
2026/09/24 22:07:06 dialing:dial unix /tmp/5840-mr-e8lKxZ_E4zdfQJZYtZ5J: connect: no such file or directory
--- PASS: TestEarlyExit (8.13s)
=== RUN   TestCrashWorker
Coordinator state changed to ReducingState
Coordinator state changed to DoneState
--- PASS: TestCrashWorker (62.20s)
PASS
2026/09/24 22:08:08 dialing:dial unix /tmp/5840-mr-4NvqXcDR0Tjn60nBRkag: connect: no such file or directory
2026/09/24 22:08:09 dialing:dial unix /tmp/5840-mr-4NvqXcDR0Tjn60nBRkag: connect: no such file or directory
2026/09/24 22:08:09 dialing:dial unix /tmp/5840-mr-4NvqXcDR0Tjn60nBRkag: connect: no such file or directory
ok      6.5840/mr       130.723s

以下是关于 Go 的记录：

之前试过 Go 被劝退了，正好这门课程再来尝试一下。

中间遇到了 gob 的坑，对于“零”值，传输时不会传输零值字段，因此也丕会覆盖。必须记得在每轮循环中重置 reply，否则会出现某些字段一直是上一次的值，而不是期待的“零”值。

不再赘述，直接参考以下 Gemini 的看法：

点击展开

这个问题出现的根本原因是：在 Worker 函数中，reply 变量是在 for 循环外层声明的，且每轮循环没有重置；而 Go 的 net/rpc 底层使用 encoding/gob，在序列化时会自动省略零值字段（即 TaskID == 0）。

问题定位与执行轨迹还原

从你的日志中可以看到两组关键信息：

  1. Coordinator 分配的总是 Task 0：FileName 是 ../../main/pg-dorian_gray.txt，对应 TaskID = 0。
  2. Worker 打印的 reply.TaskID 却依次是 5、2、1、7：
1
2
3
4
5
6
Assign map task 0 to worker
Worker 243838: call reply ok=true, reply={HasAssign:true TaskType:0 TaskID:5 FileName:../../main/pg-dorian_gray.txt ...}
...
Assign map task 0 to worker
Worker 243836: call reply ok=true, reply={HasAssign:true TaskType:0 TaskID:2 FileName:../../main/pg-dorian_gray.txt ...}

为什么 TaskID 会变成旧的值？

  1. encoding/gob 的零值优化机制：
    Go 标准库的 encoding/gob 在编码传输时，为了节省网络带宽，不会传输结构体中的零值字段（例如整数 0、布尔 false、空字符串 ""）。
  2. 接收端内存未清理：
    当 Coordinator 发送 TaskID: 0 时，网络包中没有包含 TaskID 的键值。Worker 端反序列化到 &reply 时，解码器只更新收到的非零字段（如 FileName、NReduce 等），而未收到的字段直接保留 reply 原结构体中的当前值。
  3. 连锁死锁反应：
  • Worker 243838 上一次执行的是 Task 5，因此 reply.TaskID 为 5。
  • 收到 Task 0 的任务后，FileName 变成了 pg-dorian_gray.txt，但 TaskID 依旧是 5。
  • Worker 跑完后向 Coordinator 发送完成汇报：Finished: true, TaskID: 5。
  • Coordinator 收到汇报，将已经完成过的 Task 5 再次标记完成，而 **Task 0 的状态一直卡在 TaskRunningState**。
  • 10 秒超时后，Coordinator 判定 Task 0 超时未完成，又将其分配给下一个空闲 Worker，再次因为 TaskID = 0 无法被解码覆盖，导致所有 Worker 轮流“替罪”，Task 0 永远无法完成。

修复方法

在 Worker 函数的 for 循环内部，**每轮发起 RPC 之前，必须重新初始化 reply**（建议连同 args 的重置逻辑一起规范）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
func Worker(sockname string, mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	coordSockName = sockname

	args := Params{}
	for {
		// 关键修复：每一轮循环都必须使用全新的 reply 结构体
		reply := Response{}

		fmt.Printf("Worker %d: call args: %+v\n", os.Getpid(), args)

		ok := call("Coordinator.Manage", &args, &reply)

		fmt.Printf("Worker %d: call reply ok=%v, reply=%+v\n", os.Getpid(), ok, reply)

		if !ok {
			log.Printf("%d: call failed", os.Getpid())
			return
		}

		if reply.HasAssign {
			if reply.TaskType == MapTask {
				// 执行 Map 任务逻辑...
				// ...
				args = Params{
					Finished: true,
					TaskID:   reply.TaskID,
					TaskType: reply.TaskType,
				}
			} else if reply.TaskType == ReduceTask {
				// 执行 Reduce 任务逻辑...
				// ...
				args = Params{
					Finished: true,
					TaskID:   reply.TaskID,
					TaskType: reply.TaskType,
				}
			}
		} else {
			// 没有拿到任务时 sleep，并清空完成参数，避免下次请求误提交
			time.Sleep(500 * time.Millisecond)
			args = Params{
				Finished: false,
			}
		}
	}
}

除此之外，Go 的 sync 相关也很方便。

以下是 Go web crawler，我记得 https://go.dev/tour/ 也有。

此程序的目标应该是抓取网页，然后递归继续抓取网页中的 URL（排除掉已经抓取过的），直到没有发现新的 URL 或者达到深度限制。

选举于维护一个 fetchedUrls map[string]bool 来记录已经抓取过的 URL，避免重复抓取，但是同时又需要保证并发安全，一般来说应该使用锁来保护这个共享资源。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
package main

import (
	"fmt"
	"sync"
)

var mu sync.Mutex
var fetchedUrls = make(map[string]bool)

type Fetcher interface {
	// Fetch returns the body of URL and
	// a slice of URLs found on that page.
	Fetch(url string) (body string, urls []string, err error)
}

func testAndSet(url string) bool {
	mu.Lock()
	defer mu.Unlock()

	if fetchedUrls[url] {
		return true
	}

	fetchedUrls[url] = true
	return false
}

// Crawl uses fetcher to recursively crawl
// pages starting with url, to a maximum of depth.
func Crawl(url string, depth int, fetcher Fetcher, wg *sync.WaitGroup) {
	// TODO: Fetch URLs in parallel.
	// TODO: Don't fetch the same URL twice.
	// This implementation doesn't do either:
	if depth <= 0 {
		return
	}

	if testAndSet(url) {
		return
	}

	body, urls, err := fetcher.Fetch(url)
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Printf("found: %s %q\n", url, body)

	for _, u := range urls {
		wg.Go(func() {
			Crawl(u, depth-1, fetcher, wg)
		})
	}
}

func main() {
	wg := sync.WaitGroup{}
	wg.Go(func() {
		Crawl("https://golang.org/", 4, fetcher, &wg)
	})
	wg.Wait()
}

// fakeFetcher is Fetcher that returns canned results.
type fakeFetcher map[string]*fakeResult

type fakeResult struct {
	body string
	urls []string
}

func (f fakeFetcher) Fetch(url string) (string, []string, error) {
	if res, ok := f[url]; ok {
		return res.body, res.urls, nil
	}
	return "", nil, fmt.Errorf("not found: %s", url)
}

...

Go 的 defer 非常有意思，对于对整个函数作用域生效的锁，再也不用对着每个 return 写 mu.Unlock() 了，显然还可以有很多用法，文件释放之类的。

此外 WaitGroup 也很方便，wg.Add(1) 增加计数，wg.Done() 减少计数，wg.Wait() 阻塞等待计数归零。可以当作一个计数信号量来使用，利用 wg.Go(func() { ... }) 来启动一个 goroutine 很顺手，go 本身可以拿来启动 goroutine，很简洁。虽然探索的还不多，但相比之前在 Python 写并发的经历，显然 Go 的并发设计很有一套，确实 go。

以上使用 mutex，然而 Go 还有 channel 原语：

Channels are a typed conduit through which you can send and receive values with the channel operator, <-.

Channels 是一种类型化的管道，通过 channel 操作符 <- 可以发送和接收值。

1
2
3
ch <- v    // Send v to channel ch.
v := <-ch  // Receive from ch, and
           // assign value to v.

(The data flows in the direction of the arrow.)

数据流动的方向与箭头一致。

Like maps and slices, channels must be created before use:

和 maps 和 slices 一样，channels 必须在使用前创建：

1
ch := make(chan int)

By default, sends and receives block until the other side is ready. This allows goroutines to synchronize without explicit locks or condition variables.

默认情况下，发送和接收会阻塞，直到另一端准备好。这使得 goroutines 可以在没有显式锁或条件变量的情况下进行同步。

以上摘自 https://go.dev/tour/concurrency/2

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
package main

import (
	"fmt"
	"sync"
)

type Fetcher interface {
	// Fetch returns the body of URL and
	// a slice of URLs found on that page.
	Fetch(url string) (body string, urls []string, err error)
}

type crawlResult struct {
	depth int
	urls  []string
}

// Crawl uses fetcher to recursively crawl
// pages starting with url, to a maximum of depth.
func Crawl(url string, depth int, fetcher Fetcher) {
	// TODO: Fetch URLs in parallel.
	// TODO: Don't fetch the same URL twice.

	if depth <= 0 {
		fmt.Printf("depth limit reached for: %s\n", url)
		return
	}

	body, urls, err := fetcher.Fetch(url)
	if err != nil {
		fmt.Println(err)
		results <- crawlResult{depth: depth - 1, urls: []string{url}}
		return
	}
	fmt.Printf("found: %s %q\n", url, body)

	results <- crawlResult{depth: depth - 1, urls: urls}
}

var tasks = make(chan string, 100)
var results = make(chan crawlResult, 100)

func worker(depth int, fetcher Fetcher) {
	for url := range tasks {
		Crawl(url, depth, fetcher)
	}
}

func coordinator(url string, depth int, fetcher Fetcher) {
	fetchedUrls := make(map[string]bool)
	pendingtask := 0

	tasks <- url
	pendingtask++
	fetchedUrls[url] = true

	worker_wg := sync.WaitGroup{}

	for range 4 {
		worker_wg.Go(func() {
			worker(depth, fetcher)
		})
	}

	for pendingtask > 0 {
		// Process results
		result := <-results
		pendingtask--

		if result.depth <= 0 {
			continue
		} else {
			for _, url := range result.urls {
				if !fetchedUrls[url] {
					tasks <- url
					fetchedUrls[url] = true
					pendingtask++
				}
			}
		}
	}

	close(tasks)

	worker_wg.Wait()

	close(results)
}

func main() {
	coordinator("https://golang.org/", 4, fetcher)
}

// fakeFetcher is Fetcher that returns canned results.
type fakeFetcher map[string]*fakeResult

type fakeResult struct {
	body string
	urls []string
}

func (f fakeFetcher) Fetch(url string) (string, []string, error) {
	if res, ok := f[url]; ok {
		return res.body, res.urls, nil
	}
	return "", nil, fmt.Errorf("not found: %s", url)
}

...

将 fetchedUrls 的读写直接全部留在无竞争的 coordinator 中，worker 只负责抓取网页并返回结果，由 channel 通信来交分配任务交换结果，这样不用锁确实很优雅~

我确实觉得脑洞大开，但是 channel 的实现呢？说不定锁只是跑到了后面…离理解似乎还差太远了

TODO：

天啊这次的 TODO 居然要列表哈哈哈哈…我这辈子会点开来看吗，真绝望

说些什么吧！

giscus