Golang編程并發(fā)工具庫(kù)MapReduce使用實(shí)踐

更新時(shí)間：2022年04月29日 09:50:59 作者：EvaCcino

這篇文章主要為大家介紹了Golang并發(fā)工具庫(kù)MapReduce的使用實(shí)踐，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

環(huán)境

go version go1.16.4 windows/amd64
Intel(R) Core(TM) i7-7820HK CPU @ 2.90GHz 4核心8線程

項(xiàng)目需求

處理數(shù)個(gè)約5MB的小文件

從源目錄讀取文件并拷貝到目標(biāo)目錄

計(jì)算源文件MD5和目標(biāo)文件MD5進(jìn)行對(duì)比，如不相同則報(bào)錯(cuò)并終止程序執(zhí)行

mapReduce使用說(shuō)明

go get -u github.com/tal-tech/go-zero

需求實(shí)現(xiàn)

判斷上下文是否中止 → 讀取數(shù)據(jù) → 寫(xiě)入數(shù)據(jù) → 校驗(yàn)MD5

 func fnBuilder(name string) func() error {
   return func() error {
        // 判斷上下文是否終止
        select {
        case <-ctx.Done():
          return ctx.Err()
        default:
        }
        // 讀取源數(shù)據(jù)
        data, _err := os.ReadFile(filepath.Join(sourcePath, fileName))
        // 計(jì)算源數(shù)據(jù)MD5
        sourceMD5 := hash.Md5(data)
        // 獲取名稱
        fields := strings.Split(d.Name(), "-")
        // 目標(biāo)文件路徑
        distFilePath := filepath.Join(distPath, fileName)
        // 拷貝數(shù)據(jù)
        os.WriteFile(distFilePath, data, 0600)
        // 校驗(yàn)數(shù)據(jù)
        distData, _err := os.ReadFile(distFilePath)
        distMD5 := hash.Md5(distData)
        if !bytes.EqualFold(sourceMD5, distMD5) {
          return errors.New("md5校驗(yàn)失敗")
        }
        return nil
      }
 }

業(yè)務(wù)邏輯

創(chuàng)建任務(wù)隊(duì)列

type SourceMap = map[string]fs.DirEntry
func CopyFileToDist(ctx context.Context, source SourceMap) (err error) {
  // 創(chuàng)建工作隊(duì)列
  work := make([]func() error, 0, len(source))
  for _name := range source {
    // 創(chuàng)建任務(wù)
    work = append(work, fnBuilder(_name))
  }
  switch concurrency {
  default:
    // mapReduce
  case 1:
    // sync.waitGroup
  case 2:
    // 串行
  }
}

執(zhí)行方式1：MapReduce

 func() {
    if err = mr.Finish(work...); err != nil {
      return err
    }
 }

執(zhí)行方式2：sync.WaitGroup

func() {
    var wg sync.WaitGroup
    wg.Add(len(work))
    for k := range work {
      go func(index int) {
        defer wg.Done()
        if err = work[index](); err != nil {
          log.Errorln(err)
          return
        }
      }(k)
    }
    wg.Wait()
}

執(zhí)行方式3：串行

func() {
    for _, fn := range work {
      if err = fn(); err != nil {
        return err
      }
    }
}

運(yùn)行結(jié)果

MapReduce

耗時(shí) 109220900 ns

{"file":"D:/go/src/filenamesSorter/main.go:44","func":"main.init.0","level":"info","msg":"并發(fā)處理（0-mapReduce 1-Sync.WaitGroup 2-不并發(fā)） 0","time":"2021-06-02T13:32:05+08:00"}
{"file":"D:/go/src/filenamesSorter/main.go:69","func":"main.main","level":"info","msg":"文件分類(lèi)完畢","time":"2021-06-02T13:32:05+08:00","文件數(shù)":17,"耗時(shí)(ns)":109220900}

sync.WaitGroup

耗時(shí) 109798000 ns

{"file":"D:/go/src/filenamesSorter/main.go:44","func":"main.init.0","level":"info","msg":"并發(fā)處理（0-mapReduce 1-Sync.WaitGroup 2-不并發(fā)） 1","time":"2021-06-02T13:31:28+08:00"}
{"file":"D:/go/src/filenamesSorter/main.go:69","func":"main.main","level":"info","msg":"文件分類(lèi)完畢","time":"2021-06-02T13:31:28+08:00","文件數(shù)":17,"耗時(shí)(ns)":109798000}

串行

耗時(shí) 359307700 ns

{"file":"D:/go/src/filenamesSorter/main.go:44","func":"main.init.0","level":"info","msg":"并發(fā)處理（0-mapReduce 1-Sync.WaitGroup 2-不并發(fā)） 2","time":"2021-06-02T13:33:02+08:00"}
{"file":"D:/go/src/filenamesSorter/main.go:69","func":"main.main","level":"info","msg":"文件分類(lèi)完畢","time":"2021-06-02T13:33:02+08:00","文件數(shù)":17,"耗時(shí)(ns)":359307700}