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

Go  /  管理员 发布于 7年前   288

业务背景

业务需求要求开发一个异步上传文件的接口，并支持上传进度的查询。

需求分析

ZIP压缩包中，包含一个csv文件和一个图片文件夹，要求：解析csv数据存入mongo，将图片文件夹中的图片信息对应上csv中的人员信息。

ZIP压缩包解压

使用golang自带的 "archive/zip" 包解压。

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}

在这个解压的过程中，压缩包的树结构只能到2层

import.zip┝┅┅import.csv┖┅┅images

在解压后，所有的文件都会在同一个目录下，既images中的图片会变成和.csv文件同级

验证csv文件编码格式是否为UTF-8

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}

后续支持utf-8转码

这个utf8编码判断方法是网上down下来的，后续优化一下

主逻辑

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 informationfunc 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)  ・・・}

将错误信息以channel接收，使用 package "sync" 的 sync.WaitGroup 控制异步协程。在入库的过程中，将当前的进度存入redis。

查询进度接口

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}

从redis中取出来的数据是[]uint8类型数据，先断言，然后转类型返回。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。