文章目录

前言

该实验的🔗：6.5840 Lab 1: MapReduce

2023年的叫做6.5840，名字改了，不过区别不大，直接用这个就行，大家还是习惯称呼他6.824。

我是根据课程表来的，所以上完了前四节课就开始做实验了。下面就正式开始吧。

1. 前置准备

  • 阅读MapReduce这篇论文，至少你得知道他的原理；
  • 对Go有基本的了解，可以看一下我另外两篇关于Go的语法学习，我两天速通了一下；
  • 配置好GO的环境，并找一个适合的编辑器，我这里用的是VSCode，进行远程连接，唯一的坏处就是同一文件夹下不能有多个package main，不然会报错，看得我很难受，翻遍了全网也不知道怎么修改，最后是选择把插件给关了，从根源上解决问题。

1.1 非分布式的实现

官方给我们提供了一个简单的顺序 mapreduce 实现。它在一个进程中一次运行一个映射和还原。
在这里插入图片描述
这个wc.go定义了Map和Reduce函数，这两个函数在mrsequential.go被加载和调用。

mrsequential.go 会在 mr-out-0 文件中留下输出结果。输入来自名为 pg-xxx.txt 的文本文件。
在这里插入图片描述

1.2 your job

我们的任务是实现一个分布式 MapReduce，它由两个程序组成：协调程序coordinator和工作程序worker。一个coordinator,一个worker（启动多个）,在这次实验都在一个机器上运行。工作进程将通过 RPC 与协调程序对话。每个 Worker 进程都会向协调器请求任务，从一个或多个文件中读取任务输入，执行任务，并将任务输出写入一个或多个文件。协调器应该注意到某个 Worker 是否超时，若超时将任务重新交给不同的 Worker。

协调器和 Worker 的 "主 "例程位于 main/mrcoordinator.go 和 main/mrworker.go；不要修改这些文件。我们要写的代码就是mr文件夹下的：mr/coordinator.go、mr/worker.go 和 mr/rpc.go 。

当我们完成后，执行bash test-mr.sh，若我们成功了，就会出现这样的代码：

$ bash test-mr.sh
*** Starting wc test.
--- wc test: PASS
*** Starting indexer test.
--- indexer test: PASS
*** Starting map parallelism test.
--- map parallelism test: PASS
*** Starting reduce parallelism test.
--- reduce parallelism test: PASS
*** Starting job count test.
--- job count test: PASS
*** Starting early exit test.
--- early exit test: PASS
*** Starting crash test.
--- crash test: PASS
*** PASSED ALL TESTS
$

所以说，我们现在执行是没有用的。

1.3 Some rules

下面的翻译来自DeepL

  • 映射阶段应将中间键分成多个桶，供 nReduce 减缩任务使用，其中 nReduce 是减缩任务的数量，也就是 main/mrcoordinator.go 传递给 MakeCoordinator() 的参数。每个映射器应创建 nReduce 中间文件，供还原任务使用。
  • Worker 实现应将第 X 个还原任务的输出放到 mr-out-X 文件中。
  • mr-out-X文件应包含每个还原函数输出的一行。该行应该以 Go "%v %v" 格式生成，并以键和值调用。请查看 main/mrsequential.go 中注释为 "这是正确格式 "的一行。如果您的实现与此格式偏差过大，测试脚本就会失败。
  • 您可以修改 mr/worker.go、mr/coordinator.go 和 mr/rpc.go。您可以临时修改其他文件进行测试，但请确保您的代码能在原始版本下运行；我们将使用原始版本进行测试。
  • Worker 应将中间 Map 输出放到当前目录下的文件中，这样 Worker 日后就可以将它们作为 Reduce 任务的输入进行读取。
  • main/mrcoordinator.go 希望 mr/coordinator.go 实现一个 Done() 方法，当 MapReduce 作业完全完成时返回 true；此时，mrcoordinator.go 将退出。
  • 当作业完全完成时，工作进程也应退出。实现这一点的简单方法是使用 call()的返回值：如果 Worker 无法与协调器取得联系，它可以认为协调器已经退出，因为作业已经完成，所以 Worker 也可以终止。根据您的设计，您可能还会发现，协调者向工人下达一个 "请退出 "的伪任务也很有帮助。

1.4 提示Hints

  • 指导页面有一些关于开发和调试的提示。
  • 一种入门方法是修改 mr/worker.go 的 Worker() 向协调器发送 RPC，请求任务。然后修改协调器，以尚未启动的Map任务的文件名作为回应。然后修改 Worker，读取该文件并调用应用程序的 Map 函数，如 mrsequential.go 所示。
  • 应用程序的 Map 和 Reduce 函数是在运行时使用 Go 插件包从名称以 .so 结尾的文件中加载的。
  • 如果更改了 mr/ 目录中的任何内容，可能需要重新构建使用的 MapReduce 插件，如 go build -buildmode=plugin …/mrapps/wc.go
  • 本实验室依赖于 Worker 共享一个文件系统。当所有 Worker 都在同一台机器上运行时，共享文件系统很简单，但如果 Worker 运行在不同的机器上，则需要类似 GFS 的全局文件系统。
  • 中间文件的合理命名约定是 mr-X-Y，其中 X 是映射任务编号，Y 是还原任务编号。
  • Worker 的 Map 任务代码需要一种方法来将中间键/值对存储到文件中，以便在 reduce 任务中正确读回。一种方法是使用 Go 的编码/json 包。将 JSON 格式的键/值对写入开放文件
  • 使用 Go 的竞赛检测器，即 go run -race。test-mr.sh 开头的注释会告诉你如何使用 -race 运行它。我们在给你的实验打分时，不会使用竞赛检测器。不过，如果您的代码存在竞赛，即使不使用竞赛检测器，我们测试时也很有可能会失败。
  • 如果您选择执行备份任务（第 3.6 节），请注意我们会测试您的代码是否会在工人执行任务而不崩溃时调度无关任务。备份任务应在一段相对较长的时间（如 10 秒）后才调度。

挑了部分我觉得比较重要的，放在这，没有放全，读者可以去我最上面的链接打开看看。

1.5 先认识下整体逻辑

  1. 将输入拆分成很多片
  2. 通过Map函数生成键值对
  3. 将键值对追加到中间结果切片intermediate 中
  4. Reduce函数将中间结果切片统计，得到最终结果
    在这里插入图片描述

2. Lab1 - MapReduce

2.1 源码解读

通读了一下mrsequential.go、wc.go这两个文件，还是蛮简单的。大致思路就是将每一个文件按照单词切分，将每一个文件的切片都追加到intermediate里面（所有的数据放在同一个place，而不是N✖M这样的buckets）；再用sort.Sort()进行排序（我觉得这是一个很新奇的点，首先他必须是一个接口，其次他必须实现三个方法才能使用：Len()、Swap(i, j int)、Less(i, j int) bool）；最后将相同Key的计算总和为多少，写入文件。

// for sorting by key.
type ByKey []mr.KeyValue	// KeyValue是一个结构体：key, value string

// for sorting by key.	sort.Sort必须实现的三个方法
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }
.
intermediate := []mr.KeyValue{}	// {}表示一个空的切片字面量
.
sort.Sort(ByKey(intermediate))	//

最下面的排序函数中，必须要有ByKey()，因为原intermediate是一个切片，但该切片并没有实现sort.Interface接口所需的方法之一——Len方法。相当于一个多态的实现吧，原本的Sort是父类，我们自定义排序，就需要把那些虚函数在子类实现了，传子类参数，才能实现。

我们先来看一下sort.Sort()的源码

type Interface interface {
  // Len is the number of elements in the collection.
  Len() int
  // Less reports whether the element with
  // index i should sort before the element with index j.
  Less(i, j int) bool
  // Swap swaps the elements with indexes i and j.
  Swap(i, j int)
}

func Sort(data Interface) {
  n := data.Len()
  quickSort(data, 0, n, maxDepth(n))
}

可见，要使用Sort，必须要实现三个方法。在某些情况下，ByKey 和[]mr.KeyValue可以等价使用，例如声明变量和函数返回值。但是，在赋值、比较、传参时，编译器会严格要求切片类型，所以需要进行转换。我们的切片是无法实现方法的，必须用type变成自己的自定义类型才可以。

2.2 思路

昨天晚上问了一个学姐，做这个的思路和想法是啥，她告诉我怼着那几个hints 去看。

首先要明确一点，我们启动mrcoordinator.go后再启动mrworker.go，前者是Master，分配任务给Worker，Worker做完任务后，通知Master进行Reduce。为了统一，后文统一说成coordinator。

Hints第一条告诉我们，修改Worker()，向coordinator发送rpc请求，请求任务分配。
在这里插入图片描述
前面我们启动mrcoordinator.go，就已经通过调用MakeCoordinator启动了服务器，监听着外来请求；上图的这一行代码，会向已经开启的监听套接字发出请求，建立连接（在作者实现的call函数中）。c, err := rpc.DialHTTP("unix", sockname)得到的c是一个RPC客户端对象，他有一个方法c.all()可以调用远程的方法即Example，做出应答，将args+1传给reply，并打印出来，说明连接成功。
在这里插入图片描述
以上就是coordinator与worker一个简单的rpc交互。

由于对外执行任务的接口只有一个worker函数，那么RPC怎么知道你要执行什么任务呢？通过上面的例子也能有一个思路，那就是在rpc中定义不同的状态，类似于有限状态机，来执行不同的任务。目前的任务大致可以分为Map、Wait、Reduce、Done。并且，coordinator还需要创建Task，那么就需要有Task的struct，属性包括任务类别、文件切片等等。OK，思路大致有了，可以开干了。由于我看那个example都是写在rpc.go的，那我也写到这里吧。

2.3 制作Map

一口气写到了晚上9点半，把第一个点子——Map部分写出来了，赶紧来记录一下。

我觉得最难的部分就是结构体的定义了。这里我先说一下我定义这个结构体的历程，我相信这部分你要是做出来了，几个函数就很简单了。当然，这或许不是最后的最终定义，或许后面做到别的，又会有更多的定义加上去，想要一步登天不太可能。

首先，一个Task结构体要有对应的File；其次，我们的Worker在取任务的时候，怎么知道这个任务是Map还是Reduce呢，所以还得有一个TaskType任务状态。

然后是Coordinator那边，作为Master，得接收文件有哪些，所以需要一个文件切片；根据hints来看，我们一开始就要定义Reduce的数量，所以这个也要有；我们的任务都是放在Coordinator的，Worker在取任务的时候，相当于是在访问公共资源，加锁太麻烦了，直接用channel，它本身是并发安全的，使用起来非常方便。

目前结构体的定义：

type Task struct {
	TaskType int	// 任务状态：Map、Reduce
	FileName string	// 文件切片
	TaskId	int		// 任务ID，生成中间文件要用
	ReduceNum int	// Reduce的数量
}

type Coordinator struct {
	// Your definitions here.
	State int	// Map Reduce阶段
	MapChan chan *Task	// Map任务channel
	ReduceChan chan *Task	// Reduce任务channel
	ReduceNum int	// Reduce的数量
	Files	[]string	// 文件
}

为了做测试，我就只是先获取了一个任务，做了Map。代码很简单，可以参考mrcoordinator.go。

work.go

func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
	// alive := true
	// for alive {
	// 	task := GetTask()
	// 	switch task.TaskType {
	// 	case MapTask: {
	// 			DoMapTask(task, mapf)	// 执行map任务
	// 		}
	// 	}
	// }
	task := GetTask()
	DoMapTask(&task, mapf)
}
// 获取任务
func GetTask() Task {
	args := TasskArgs{}	// 为空
	reply := Task{}
	if ok := call("Coordinator.PullTask", &args, &reply); ok {
		fmt.Printf("reply TaskId is %d\n", reply.TaskId)
	} else {
		fmt.Printf("call failed!\n")
	}
	return reply
}

// Map的制作
func DoMapTask(task *Task, mapf func(string, string) []KeyValue) {
	intermediate := []KeyValue{}
	fmt.Println(task.FileName)
	file, err := os.Open(task.FileName)
	if err != nil {
		log.Fatalf("cannot open %v", task.FileName)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", task.FileName)
	}
	file.Close()
	reduceNum := task.ReduceNum
	intermediate = mapf(task.FileName, string(content))
	HashKv := make([][]KeyValue, reduceNum)
	for _, v := range(intermediate) {
		index := ihash(v.Key) % reduceNum
		HashKv[index] = append(HashKv[index], v)	// 将该kv键值对放入对应的下标
	}
	// 放入中间文件
	for i := 0; i < reduceNum; i++ {
		filename := "mr-tmp-" + strconv.Itoa(task.TaskId) + "-" + strconv.Itoa(i)
		new_file, err := os.Create(filename)
		if err != nil {
			log.Fatal("create file failed:", err)
		}
		enc := json.NewEncoder(new_file)	// 创建一个新的JSON编码器
		for _, kv := range(HashKv[i]) {
			err := enc.Encode(kv)
			if err != nil {
				log.Fatal("encode failed:", err)
			}
		}
		new_file.Close()
	}
}

coordinator.go

func (c *Coordinator) PullTask(args *ExampleArgs, reply *Task) error {
	*reply = *<-c.MapChan
	return nil
}
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{State: 0, 
					MapChan: make(chan *Task, len(files)),
					ReduceChan: make(chan *Task, nReduce),
					ReduceNum: nReduce,
					Files: files}
	// 制造Map任务
	c.MakeMapTasks(files)

	c.server()	// 启动RPC服务器
	return &c
}

// 将生成的任务放入map管道
func (c *Coordinator) MakeMapTasks(files []string) {
	for id, v := range(files) {
		// 生成任务
		task := Task {TaskType: MapTask,
					FileName: v,
					TaskId: id,
					ReduceNum: c.ReduceNum}
		c.MapChan <- &task	// 写入通道

		fmt.Println(v, "写入成功！")
	}
}

在这里插入图片描述

2.4 coordinator感知各个Task运行完毕，转为Reduce任务

思来想去，Worker做完以后，要干嘛呢？无非是等待所有的Task都从Map变为Reduce，这个时候coordinator才能进行下一步。这个状态的切换和监控没必要让worker知道，所以让coordinator来判断，因此又需要引入一个新的变量，用来维护任务的状态等信息，coordinator会去check这个变量，从而知道什么时候转变为Reduce阶段。

写了一个上午+半个下午，我来梳理一下现在的情况。

目前除了Reduce任务没有写以外，其他基本都做了，在测试的时候，我依旧只做了一个Map，然后让Worker去取，做完了之后，进行任务转变——变为WaitingTask阶段，每一次去取任务的时候，都会先判断channel里面是否还有任务，若还有，则继续制作；若没有了，则需要判断是否所有的任务都进入WaitingTask阶段了，若是，则大步迈进下一个任务阶段——Reduce，否则就说明还在制作中，仍需要等待。

下面我就粘贴一下新写的一些函数。

worker.go

func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
	alive := true
	for alive {
		task := GetTask()
		switch task.TaskType {
			case MapTask: {
				DoMapTask(&task, mapf)	// 执行map任务
				TaskDone(&task)
			}
			case WaitingTask: {
				fmt.Println("All tasks are in progress, please wait...")
				time.Sleep(time.Second)
			}
			case ReduceTask: {
				DoReduceTask(&task, reducef)
			}
			case ExitTask: {
				fmt.Println("All tasks are in progress, Worker exit")
				alive = false
			}
		}
	}
}

func TaskDone(task *Task) {
	args := task
	reply := Task{}
	ok := call("Coordinator.MarkDone", &args, &reply)
	if ok {
		fmt.Println("Task Done!")
	}
}

coordinator.go

结构体的定义：

type TaskMetaInfo struct {
	TaskAddr *Task	// 任务指针 
}

// 保存全部任务的元数据
type TaskMetaHolder struct {
	MetaMap map[int]*TaskMetaInfo	
}

type Coordinator struct {
	// Your definitions here.
	State int	// Map Reduce阶段
	MapChan chan *Task	// Map任务channel
	ReduceChan chan *Task	// Reduce任务channel
	ReduceNum int	// Reduce的数量
	Files	[]string	// 文件
	taskMetaHolder TaskMetaHolder	// 任务信息
}

2.5 Reduce任务的制作

Reduce任务的制作需要等所有的任务都完成后，才能制作。制作的任务，放入ReduceChan中，做到这了就没什么难度了。
worker.go

func DoReduceTask(task *Task, reducef func(string, []string) string) {
	reduceNum := task.TaskId
	intermediate := shuffle(task.FileName)
	finalName := fmt.Sprintf("mr-out-%d", reduceNum)
	ofile, err := os.Create(finalName)
	if err != nil {
		log.Fatal("create file failed:", err)
	}
	for i := 0; i < len(intermediate); {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {	// i和j之间是一样的键值对，将一样的到一个values中
			values = append(values, intermediate[k].Value)
		}
		output := reducef(intermediate[i].Key, values)
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)
		i = j
	}
	ofile.Close()
}

// 排序，将reduce任务下的全部放一起
func shuffle(files []string) []KeyValue {
	kva := []KeyValue{}
	for _, fi := range files {
		file, err := os.Open(fi)
		if err != nil {
			log.Fatalf("cannot open %v", fi)
		}
		dec := json.NewDecoder(file)
		for {
			kv := KeyValue{}
			if err := dec.Decode(&kv); err != nil {
				break
			}
			kva = append(kva, kv)
		}
		file.Close()
	}
	sort.Sort(ByKey(kva))
	return kva
}

coordinator.go

func (t *TaskMetaHolder) checkAllTasks(state int) bool {
	UnDoneNum, DoneNum := 0, 0
	if state == MapState {
		for _, v := range t.MetaMap {
			if v.TaskAddr.TaskType == MapTask {
				UnDoneNum++
			} else if v.TaskAddr.TaskType == WaitingTask {
				DoneNum++
			}
		}
	} else {	// ReduceState
		for _, v := range(t.MetaMap) {
			if v.TaskAddr.TaskType == ReduceState {
				UnDoneNum++
			} else if v.TaskAddr.TaskType == WaitingTask {
				DoneNum++
			}
		}
	}
	if UnDoneNum == 0 && DoneNum > 0 {
		return true
	} 
	return false
}

func (c *Coordinator) ToNextState() {
	if c.State == MapState {
		c.makeReduceTasks()
		c.State = ReduceState
		fmt.Println("State changed: ReduceState")
	} else if c.State == ReduceState {
		c.State = AllDone
	}
}

func (c *Coordinator) makeReduceTasks() {
	for i := 0; i < c.ReduceNum; i++ {
		id := i + len(c.Files)
		task := Task{
			TaskType: ReduceTask,
			FileName: selectReduceFiles(i),
			TaskId: id,
		}
		taskMetaInfo := TaskMetaInfo{TaskAddr: &task}
		c.taskMetaHolder.MetaMap[id] = &taskMetaInfo	// 文件名长度后面是新的Reduce任务坐标
		c.ReduceChan <- &task
	}
}

func selectReduceFiles(reduceNum int) []string {
	s := []string{}
	path, _ := os.Getwd()	// 当前工作目录
	files, _ := ioutil.ReadDir(path)
	for _, f := range files {
		if strings.HasPrefix(f.Name(), "mr-tmp-") && strings.HasSuffix(f.Name(), strconv.Itoa(reduceNum)) {
			s = append(s, f.Name())
		}
	}
	return s
}

感觉好多代码都是冗余的，后面再精炼一下代码。

测试：
在这里插入图片描述 在这里插入图片描述
和官方的mrsequential.go对比了一下，得到了一致的结果。
在这里插入图片描述
官方给出了一个测试脚本test-mr.sh，但是我在执行的时候卡住了。于是回去再看了一下Lab文档，其中有一条提到了：If you choose to implement Backup Tasks (Section 3.6), note that we test that your code doesn’t schedule extraneous tasks when workers execute tasks without crashing. Backup tasks should only be scheduled after some relatively long period of time (e.g., 10s).

意思是：如果您选择执行备份任务（第 3.6 节），请注意我们会测试您的代码是否会在工作者执行任务而不崩溃的情况下安排无关的任务。备份任务只能在一段相对较长的时间（如 10 秒）后安排。

综上要解决的就是：
1.是否和mr-sequential产生一样的结果

2.worker是否并行执行

3.如果worker crash，你的实现能否recover

为了解决这样的Crash，就还需要给每一个任务状态加一个时钟，开一个goroutine去不断check是否有超时的Task，如果有，就需要将他放回channel。

做到这发现实现不了，似乎必须还得给每一个任务加一个Working/Waiting状态才行，不然你没法计时，太遗憾了，我得大刀阔斧改代码了。
coordinator.go

// 这个函数用来更正工作状态、修改工作超时时间
func (t *TaskMetaHolder) judgeState(taskId int) bool {
	taskInfo, ok := t.MetaMap[taskId] 
	if !ok || taskInfo.state != Waiting {
		return false	// 不用修改
	}
	taskInfo.state = Working
	taskInfo.BeginTime = time.Now()
	return true
}
// 只有所有任务都处于Done才说明执行完毕
func (t *TaskMetaHolder) checkAllTasks() bool {
	UnDoneNum, DoneNum := 0, 0
	for _, v := range t.MetaMap {
		if v.state == Done {
			DoneNum++
		} else {
			UnDoneNum++
		}
	}
	if DoneNum > 0 && UnDoneNum == 0 {
		return true
	} 
	return false
}

func (c *Coordinator) CheckTimeOut() {
	for {
		time.Sleep(2 * time.Second)	// 每2s检查一次
		mu.Lock()	// 因为要修改公共资源，需要加锁
		if c.State == AllDone {
			mu.Unlock()
			break
		}

		for _, v := range c.taskMetaHolder.MetaMap {
			if v.state == Working && time.Since(v.BeginTime) > 10*time.Second {	// 超时了
				// fmt.Printf("the task[ %d ] is crash,take [%d] s\n", v.TaskAddr.TaskId, time.Since(v.BeginTime))
				if v.TaskAddr.TaskType == MapTask {
					v.state = Waiting
					c.MapChan <- v.TaskAddr
				} else if v.TaskAddr.TaskType == ReduceTask {
					v.state = Waiting
					c.ReduceChan <- v.TaskAddr
				}
			}
		}
		mu.Unlock()
	}
}
// 任务状态的切换
func (c *Coordinator) MarkDone(args *Task, reply *Task) error {
	mu.Lock()
	defer mu.Unlock()
	meta, ok := c.taskMetaHolder.MetaMap[args.TaskId]
	if ok && meta.state == Working {
		meta.state = Done
	} else {
		// fmt.Printf("the task Id[%d] is finished,already ! ! !\n", args.TaskId)
	}
	return nil
}

最终在关闭所有I/O输出后，也是成功通过了测试！！！

太尴尬了，结束语我都写一半了，我还在炫耀能不加锁尽量不加锁，我一次通过，然后我突然想起go有一个-race功能，可以check是否有竞态情况，然后我才发现，我测试脚本的RACE=-race没开。打开再测试，Found 6 data race(s)，我直接裂开。

这次是真的成啦！起初是没有在PullTask函数加锁，加上锁就好了。我起初是觉得channel本身就是安全的了，不用再在外面加锁了。但是开启多个worker的时候，还是会有共享资源的访问啊（TaskMetaInfo.state），所以还是得加上。

在这里插入图片描述

3.Summary

这个Lab花了足足两天半完成，基本是从10点到晚上11点左右，论天数来说不算多，但总时间加起来还是蛮多的了。最开始看了网课+Go语法，然后花了半天理清思路，当时看到这个Lab的时候，眼前一亮，这才叫Lab啊，我欣喜若狂，MIT的学生能在在校期间做到如此有意思的Lab，而国内就不多说了，通过这个也能看出我国和老美计算机方面的差距。

正如网上说到的那样，第一个Lab并不难，我觉得相对较难的部分就是结构体的定义了，我做这个Lab的时候一开始也是毫无思路，然后就去看了下别的博主的一些想法，然后看了下他们对结构体的定义，再加上我自己的思考一步步来的，此话怎讲呢，如果你看完了我的这一篇，你会发现，我的结构体变量都是一步步加上去的，都是我做到某一步了，发现缺少一些实现的功能，才去加，然后大刀阔斧改代码，一开始总想着怎样精简怎么来，但后面还是得加上那些东西。

然后就是关于Hints，我觉得蛮有意思嗷，起初你觉得毫无关系的hints，到最后都用到了。就像是在玩剧本杀，每一条hints就像是一跳线索，一定指向了某一个方向，当你实现后才恍然大悟。要说这个实验的不足，我觉得，就是还可以有更多自己的思考，跳出一些博主已经实现过的方法，自己去想另一条路。

作者也说了不要公开代码，就不上传github了，我这篇文章的代码也基本罗列全了，这样也能让你有一点自己的思考。

后面会有更大的挑战等待着我，加油，这只是一个开端~

# mapreduce # 大数据 # golang # rpc
Logo
开放原子开发者工作坊

开放原子开发者工作坊旨在鼓励更多人参与开源活动，与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动，如meetup、训练营等，主打技术交流，干货满满，真诚地邀请各位开发者共同参与！

更多推荐

cover

一文解决Cellphonedb单细胞互作分析及可视化作图（2）

avatar 开放原子开发者工作坊
cover

[工具使用]——时序图与UML类图

avatar 开放原子开发者工作坊

CTFHub技能树 Web-SQL注入 详解

整数型注入我们输入 1不断尝试发现闭合方式就是 1 ，整数型存在两列order by 2存在两个注入点/?id=-1 union select 8,9爆库，当前数据库为sqli/?id=-1 union select 8,database()爆出所有表名,这里我们需要吧sqli转换为16进制/?id=-1 union select 8,group_concat(table_name) from i

avatar 开放原子开发者工作坊