Golang实现异步上传文件支持进度条查询的方法

func decompresszip(filepath, dest string) (string, string, error) {
  var csvname string
  imagefolder := path.base(filepath)
  ext := path.ext(filepath)
  foldername := strings.trimsuffix(imagefolder, ext)
  src, err := os.open(filepath)
  if err != nil {
    return "", "", err
  }
  defer src.close()

  zipfile, err := zip.openreader(src.name())
  if err != nil {
    return "", "", err
  }
  defer zipfile.close()

  err = os.mkdirall(path.join(dest, foldername), os.modeperm)
  for _, innerfile := range zipfile.file {
    info := innerfile.fileinfo()
    if info.isdir() {
      continue
    }
    dst, err := os.create(path.join(dest, foldername, info.name()))
    if err != nil {
      fmt.println(err.error())
      continue
    }
    src, err := innerfile.open()
    if err != nil {
      fmt.println(err.error())
      continue
    }
    io.copy(dst, src)
  }
  destpath, err := ioutil.readdir(path.join(dest, foldername))
  if err != nil {
    return "", "", err
  }
  for _, v := range destpath {
    if path.ext(v.name()) == ".csv" {
      csvname = path.join(dest, foldername, v.name())
    }
  }
  return foldername, csvname, nil
}

import.zip

┝┅┅import.csv

┖┅┅images

func validutf8(buf []byte) bool {
  nbytes := 0
  for i := 0; i < len(buf); i++ {
    if nbytes == 0 {
      if (buf[i] & 0x80) != 0 { //与操作之后不为0，说明首位为1
        for (buf[i] & 0x80) != 0 {
          buf[i] <<= 1 //左移一位
          nbytes++   //记录字符共占几个字节
        }
        if nbytes < 2 || nbytes > 6 { //因为utf8编码单字符最多不超过6个字节
          return false
        }
        nbytes-- //减掉首字节的一个计数
      }
    } else { //处理多字节字符
      if buf[i]&0xc0 != 0x80 { //判断多字节后面的字节是否是10开头
        return false
      }
      nbytes--
    }
  }
  return nbytes == 0
}

type linewrong struct {
  linenumber int64 `json:"line_number"`
  msg    string `json:"msg"`
}

func import(/*自定义参数*/){
  // decompress zip file to destination address
  folder, csvname, err := decompress(path.join(constant.folderprefix, req.filepath), dest)
  if err != nil {
    fmt.println(err.error())
  }

  // check if the file encoding is utf8
  b, err := ioutil.readfile(csvname)
  if err != nil {
    fmt.println(err.error())
  }
  if !utils.validutf8(b) {
    fmt.println(errors.new("数据编码错误，请使用utf-8格式csv！"))
  }

  // create goroutine to analysis data into mongodb
  var wg sync.waitgroup
  wg.add(1)

  // used to interrupt goroutine
  resultchan := make(chan error)
  // used to record wrong row in csv
  lw := make(chan []linewrong)
  go func(ctx *gin.context, name, csvpath, dir, folder string) {
    defer wg.done()
    tidt, cit, lwt, err := importcsv(ctx, name, csvpath, dir, folder)
    resultchan <- err
    if err != nil {
      fmt.println(err.error())
    }
    lw <- lwt
    if len(lwt) == 0 {
      importclassdata(ctx, tidt, cit)
    }
  }(ctx, req.name, csvname, dest, folder)

  err = <-resultchan
  linewrong := <-lw
  close(lw)
  ···
}

// pre-analysis data in csv and through wrong data with line numbers and information
func importcsv()(){
  ···
}

// analysis data again and save data into mongodb, if is there any error,through them same as import()
func importclassdata()(){
  ···
  conn, err := connect()
  if err != nil {
    return err
  }
  defer conn.close()
  conn.do("hset", taskid, "task_id", (curline*100)/totallines)
  ···
}

func queryimport()(){
  conn, err := connect()
  if err != nil {
    return nil, err
  }
  defer conn.close()

  progress, _ := conn.do("hget", key, field)
  if pro, ok := progress.([]uint8); ok {
    ba := []byte{}
    for _, b := range pro {
      ba = append(ba, byte(b))
    }
    progress,_ = strconv.atoi(string(ba))
  }
  return progress
}