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

业务背景

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

需求分析

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 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)
    ···
}

将错误信息以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类型数据，先断言，然后转类型返回。

有疑问加站长微信联系（非本文作者）