该版本代码实现了单机单进程版本，Mapper、Reducer和Combiner的实现用协程goroutine实现，通信采用channel。代码写的比较随意，没有解耦合。
功能：统计给定文件中出现的最高频的10个单词 输入：大文件 输出：最高频的10个单词 实现：5个Mapper协程、2个Reducer、1个Combiner。 为了方便起见，Combiner对最高频的10个单词进行堆排序处理，按规范来说应该放在用户程序处理。
文件目录如下，其中bin文件夹下的big_input_file.txt为输入文件，可以调用generate下的main文件生成，caller文件为入口的用户程序，master目录下分别存放master、mapper、reducer、combiner代码：
.
├── README.md
├── bin
│ └── file-store
│ └── big_input_file.txt
└── src
├── caller
│ └── main.go
├── generate
│ └── main.go
└── master
├── combiner.go
├── mapper.go
├── master.go
└── reducer.go
6 directories, 8 files
2.1 caller
用户程序，读入文件并按固定行数进行划分；然后调用master.Handle进行处理。
package main
import (
"os"
"path"
"path/filepath"
"bufio"
"strconv"
"master"
"github.com/vinllen/go-logger/logger"
)
const (
LIMIT int = 10000 // the limit line of every file
)
func main() {
curDir, err := filepath.Abs(filepath.Dir(os.Args[0]))
if err != nil {
logger.Error("Read path error: ", err.Error())
return
}
fileDir := path.Join(curDir, "file-store")
_ = os.Mkdir(fileDir, os.ModePerm)
// 1. read file
filename := "big_input_file.txt"
inputFile, err := os.Open(path.Join(fileDir, filename))
if err != nil {
logger.Error("Read inputFile error: ", err.Error())
return
}
defer inputFile.Close()
// 2. split inputFile into several pieces that every piece hold 100,000 lines
filePieceArr := []string{}
scanner := bufio.NewScanner(inputFile)
piece := 1
Outter:
for {
outputFilename := "input_piece_" + strconv.Itoa(piece)
outputFilePos := path.Join(fileDir, outputFilename)
filePieceArr = append(filePieceArr, outputFilePos)
outputFile, err := os.Create(outputFilePos)
if err != nil {
logger.Error("Split inputFile error: ", err.Error())
continue
}
defer outputFile.Close()
for cnt := 0; cnt < LIMIT; cnt++ {
if !scanner.Scan() {
break Outter
}
_, err := outputFile.WriteString(scanner.Text() + "n")
if err != nil {
logger.Error("Split inputFile writting error: ", err.Error())
return
}
}
piece++
}
// 3. pass to master
res := master.Handle(filePieceArr, fileDir)
logger.Warn(res)
}
2.2 master
Master程序，依次生成Combiner、Reducer、Mapper，处理消息中转，输出最后结果。
package master
import (
"github.com/vinllen/go-logger/logger"
)
var (
MapChanIn chan MapInput // channel produced by master while consumed by mapper
MapChanOut chan string // channel produced by mapper while consumed by master
ReduceChanIn chan string // channel produced by master while consumed by reducer
ReduceChanOut chan string // channel produced by reducer while consumed by master
CombineChanIn chan string // channel produced by master while consumed by combiner
CombineChanOut chan []Item // channel produced by combiner while consumed by master
)
func Handle(inputArr []string, fileDir string) []Item {
logger.Info("handle called")
const(
mapperNumber int = 5
reducerNumber int = 2
)
MapChanIn = make(chan MapInput)
MapChanOut = make(chan string)
ReduceChanIn = make(chan string)
ReduceChanOut = make(chan string)
CombineChanIn = make(chan string)
CombineChanOut = make(chan []Item)
reduceJobNum := len(inputArr)
combineJobNum := reducerNumber
// start combiner
go combiner()
// start reducer
for i := 1; i <= reducerNumber; i++ {
go reducer(i, fileDir)
}
// start mapper
for i := 1; i <= mapperNumber; i++ {
go mapper(i, fileDir)
}
go func() {
for i, v := range(inputArr) {
MapChanIn <- MapInput{
Filename: v,
Nr: i + 1,
} // pass job to mapper
}
close(MapChanIn) // close map input channel when no more job
}()
var res []Item
outter:
for {
select {
case v := <- MapChanOut:
go func() {
ReduceChanIn <- v
reduceJobNum--
if reduceJobNum <= 0 {
close(ReduceChanIn)
}
}()
case v := <- ReduceChanOut:
go func() {
CombineChanIn <- v
combineJobNum--
if combineJobNum <= 0 {
close(CombineChanIn)
}
}()
case v := <- CombineChanOut:
res = v
break outter
}
}
close(MapChanOut)
close(ReduceChanOut)
close(CombineChanOut)
return res
}
