Golang並行讀取檔案資料並寫入資料庫的專案實踐

2022-06-14 14:02:15

需求

最近接到一個任務，要把一批檔案中的十幾萬條JSON格式資料寫入到Oracle資料庫中，Oracle是企業級別的資料庫向來以高效能著稱，所以儘可能地利用這一特性。當時第一時間想到的就是用多執行緒並行讀檔案並運算元據庫，而Golang是為並行而生的，用Golang進行並行程式設計非常方便，因此這裡選用Golang並行讀取檔案並用Gorm運算元據庫。然而Gorm官方並不支援Oraclc，所以要藉助第三方驅動，之前寫了篇文章來記錄Gorm操作Oracle的踩坑，詳見使用Gorm操作Oracle資料庫踩坑

專案結構

data資料夾中包含數十個.out結尾的資料檔案，model.go宣告資料型別，main.go中編寫並行邏輯和資料庫操作程式碼

|——db_test
|    |——data
|        |——xxx.out
|        |——yyy.out
|    |——model
|        |——model.go
|    |——main.go
|    |——go.mod

獲取data目錄下的檔案

Golang自帶的os庫就可以對檔案、目錄進行各種豐富的操作，OpenFile函數第一個引數是目錄的路徑，第二個參數列示唯讀，第三個引數os.ModeDir表示以資料夾模式開啟。ReadDir傳入負數表示讀取目錄下所有檔案資訊，傳入n表示讀取前n個檔案資訊。最後將所有檔名儲存到字串陣列並返回。

func loadFile(path string) []string {
        // 開啟指定資料夾
    f, err := os.OpenFile(path, os.O_RDONLY, os.ModeDir)
    if err != nil {
        log.Fatalln(err.Error())
        os.Exit(0)
    }
    defer f.Close()
        // 讀取目錄下所有檔案
    fileInfo, _ := f.ReadDir(-1)

    files := make([]string, 0)
    for _, info := range fileInfo {
        files = append(files, info.Name())
    }
    return files
}

按行讀取文字資料

這裡使用bufio.Scanner來一行一行讀取JSON格式的資料，bufio.Reader也能實現按行讀取，但bufio.Scanner是go1.1後開發的模組操作起來更簡單一點。

func readRecord(filename string) {
    log.Println(filename)
    f, err := os.Open(filename)
    if err != nil {
        log.Println(filename + " error")
        return
    }
    defer f.Close()
    scanner := bufio.NewScanner(f)
    for scanner.Scan() {
        line := scanner.Text() // line就是每行文字
                // 對line進行處理
    }
}

資料型別定義

還是假設資料庫中有一個SHOPS表，結構體方法TableName指定該型別對應的資料表，編寫如下model.go檔案

package model

type ShopInfo struct {
    ShopId   string `gorm:"column:SHOPID;not null"`
    ShopName string `gorm:"column:SHOPNAME;not null"`
        // 省略剩餘的欄位
}
func (s *ShopInfo) TableName() string {
    return "SHOPS"
}

並行讀取檔案

基本邏輯是主函數讀取資料夾下面的所有檔案，然後用迴圈開啟goroutine並傳入檔名和資料庫指標，goroutine中按行讀取每個檔案並將其JSON資料轉換為結構體，在呼叫Gorm寫入Oracle資料庫。這裡用Golang的等待組來同步主函數與goroutine。

var wg sync.WaitGroup
func main() {
        // 開啟Oracle連線
    db, err := gorm.Open(oracle.Open("database/password@127.0.0.1:1521/XE"), &gorm.Config{
        Logger: logger.New(log.New(os.Stdout, "rn", log.LstdFlags), logger.Config{
            SlowThreshold: 1 * time.Millisecond,
            LogLevel:      logger.Error,
            Colorful:      true,
        }),
    })
    if err != nil {
        log.Fatalln(err)
    }

    if e := db.AutoMigrate(&model.ShopInfo{}); e != nil {
        log.Fatalln(e.Error())
    }

    path := "./data/"
    files := loadFile(path) // 載入所有檔名
        // 迴圈建立goroutine
    for i, v := range files {
        wg.Add(1)
                // 將資料庫指標和檔名傳給goroutine處理
        go writeRecord(db, path+v)
    }

    wg.Wait() // 等待所有goroutine執行完成
    log.Println("over")

}

將資料寫入資料庫

由於這些檔案中可能有重複的資料，所以這裡呼叫了Gorm的Clauses設定，當有主鍵重複的資料什麼都不做，有些情況下主鍵相同但是更新了某些欄位，這時可以用Clauses設定主鍵重複時進行更新操作。雖然主鍵重複時什麼都不做，但是db的執行結果也會包含"unique constraint"錯誤，所以在錯誤處理時要排除主鍵衝突的情況，把其他錯誤（如欄位太長或型別不匹配）記錄下來。

func writeRecord(db *gorm.DB, filename string) {
    defer wg.Done() // 不要忘記等待組-1
    f, err := os.Open(filename)
    if err != nil {
        log.Println(filename + " error")
        return
    }
    defer f.Close()
    scanner := bufio.NewScanner(f)
    iter := 0 // 記錄出錯的行數
    for scanner.Scan() {
        var shop model.ShopInfo
        iter++
        // 呼叫json.Unmarshal()將文字轉換為結構體
        if err = json.Unmarshal([]byte(scanner.Text()), &shop); err != nil {
            log.Println("轉換錯誤--->" + scanner.Text())
            return
        }
        // 用clause設定當發生ID衝突時什麼都不做
        res := db.Clauses(clause.OnConflict{DoNothing: true}).Create(&shop)
        // 雖然ID相同時程式不會停止，但是還是有錯誤返回
        // 所以這裡排除ID衝突錯誤，將其他錯誤（欄位衝突）列印出來
        if res.Error != nil && !strings.Contains(res.Error.Error(), "unique constraint") {
            log.Println("插入出錯--->" + shop.ShopId + " 在" + filename + "第" + strconv.Itoa(iter) + "行")
            return
        }
    }
}

完整main.go程式碼

將上面每一步整合後得到完整的主函數程式碼如下：

package main

import (
    "bufio"
    "db_test/model"
    "encoding/json"
    "log"
    "os"
    "strconv"
    "strings"
    "sync"
    "time"

    "github.com/cengsin/oracle"
    "gorm.io/gorm"
    "gorm.io/gorm/clause"
    "gorm.io/gorm/logger"
)

var wg sync.WaitGroup

func main() {
    log.Println("initial database connect……")
    db, err := gorm.Open(oracle.Open("database/password@127.0.0.1:1521/XE"), &gorm.Config{
        Logger: logger.New(log.New(os.Stdout, "rn", log.LstdFlags), logger.Config{
            SlowThreshold: 1 * time.Millisecond,
            LogLevel:      logger.Error,
            Colorful:      true,
        }),
    })
    if err != nil {
        log.Fatalln(err)
    }

    if e := db.AutoMigrate(&model.ShopInfo{}); e != nil {
        log.Fatalln(e.Error())
    }

    path := "../out1/"
    files := loadFile(path)
    time.Sleep(2 * time.Second)
    for i, v := range files {
        wg.Add(1)
        go writeRecord(db, path+v)
    }

    wg.Wait()
    log.Println("over")

}

func loadFile(path string) []string {
        // 開啟指定資料夾
    f, err := os.OpenFile(path, os.O_RDONLY, os.ModeDir)
    if err != nil {
        log.Fatalln(err.Error())
        os.Exit(0)
    }
    defer f.Close()
        // 讀取目錄下所有檔案
    fileInfo, _ := f.ReadDir(-1)

    files := make([]string, 0)
    for _, info := range fileInfo {
        files = append(files, info.Name())
    }
    return files
}

func writeRecord(db *gorm.DB, filename string) {
    defer wg.Done()
    f, err := os.Open(filename)
    if err != nil {
        log.Println(filename + " error")
        return
    }
    defer f.Close()
    scanner := bufio.NewScanner(f)
    iter := 0 // 記錄出錯的行數
    for scanner.Scan() {
        var shop model.ShopInfo
        iter++
        // 呼叫json.Unmarshal()將文字轉換為結構體
        if err = json.Unmarshal([]byte(scanner.Text()), &shop); err != nil {
            log.Println("轉換錯誤--->" + scanner.Text())
            return
        }
        // 用clause設定當發生ID衝突時什麼都不做
        res := db.Clauses(clause.OnConflict{DoNothing: true}).Create(&shop)
        // 雖然ID相同時程式不會停止，但是還是有錯誤返回
        // 所以這裡排除ID衝突錯誤，將其他錯誤（欄位衝突）列印出來
        if res.Error != nil && !strings.Contains(res.Error.Error(), "unique constraint") {
            log.Println("插入出錯--->" + shop.ShopId + " 在" + filename + "第" + strconv.Itoa(iter) + "行")
            return
        }
    }
}

測試執行

go run ./main.go執行過程非常快，十幾萬條資料幾分鐘就寫完了，並且CPU佔用率100%，證明非常有效的利用了並行優勢。若是檔案數量太多（上千個）的話會建立非常多goroutine，可能消耗非常多系統資源，可以在迴圈建立goroutine時進行限制，只建立30或50個，一個goroutine結束後再給它傳入一個新的檔名。

到此這篇關於Golang並行讀取檔案資料並寫入資料庫的專案實踐的文章就介紹到這了,更多相關Golang並行讀取並寫入內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！