溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇內容介紹了“怎么使用Golang并發讀取文件數據并寫入數據庫”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！
data文件夾中包含數十個.out結尾的數據文件，model.go聲明數據類型，main.go中編寫并發邏輯和數據庫操作代碼
|——db_test | |——data | |——xxx.out | |——yyy.out | |——model | |——model.go | |——main.go | |——go.mod
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, "\r\n", 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
}
}
}將上面每一步整合后得到完整的主函數代碼如下：
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, "\r\n", 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并發讀取文件數據并寫入數據庫”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
