Golang并發(fā)讀取文件數(shù)據(jù)并寫入數(shù)據(jù)庫(kù)的項(xiàng)目實(shí)踐
需求
最近接到一個(gè)任務(wù)，要把一批文件中的十幾萬(wàn)條JSON格式數(shù)據(jù)寫入到Oracle數(shù)據(jù)庫(kù)中，Oracle是企業(yè)級(jí)別的數(shù)據(jù)庫(kù)向來(lái)以高性能著稱，所以盡可能地利用這一特性。當(dāng)時(shí)第一時(shí)間想到的就是用多線程并發(fā)讀文件并操作數(shù)據(jù)庫(kù)，而Golang是為并發(fā)而生的，用Golang進(jìn)行并發(fā)編程非常方便，因此這里選用Golang并發(fā)讀取文件并用Gorm操作數(shù)據(jù)庫(kù)。然而Gorm官方并不支持Oraclc，所以要借助第三方驅(qū)動(dòng)，之前寫了篇文章來(lái)記錄Gorm操作Oracle的踩坑，詳見使用Gorm操作Oracle數(shù)據(jù)庫(kù)踩坑
項(xiàng)目結(jié)構(gòu)
data文件夾中包含數(shù)十個(gè).out結(jié)尾的數(shù)據(jù)文件，model.go聲明數(shù)據(jù)類型，main.go中編寫并發(fā)邏輯和數(shù)據(jù)庫(kù)操作代碼
|——db_test | |——data | |——xxx.out | |——yyy.out | |——model | |——model.go | |——main.go | |——go.mod
獲取data目錄下的文件
Golang自帶的os庫(kù)就可以對(duì)文件、目錄進(jìn)行各種豐富的操作，OpenFile函數(shù)第一個(gè)參數(shù)是目錄的路徑，第二個(gè)參數(shù)表示只讀，第三個(gè)參數(shù)os.ModeDir表示以文件夾模式打開。ReadDir傳入負(fù)數(shù)表示讀取目錄下所有文件信息，傳入n表示讀取前n個(gè)文件信息。最后將所有文件名保存到字符串?dāng)?shù)組并返回。
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
}按行讀取文本數(shù)據(jù)
這里使用bufio.Scanner來(lái)一行一行讀取JSON格式的數(shù)據(jù)，bufio.Reader也能實(shí)現(xiàn)按行讀取，但bufio.Scanner是go1.1后開發(fā)的模塊操作起來(lái)更簡(jiǎn)單一點(diǎn)。
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就是每行文本
// 對(duì)line進(jìn)行處理
}
}
數(shù)據(jù)類型定義
還是假設(shè)數(shù)據(jù)庫(kù)中有一個(gè)SHOPS表，結(jié)構(gòu)體方法TableName指定該類型對(duì)應(yīng)的數(shù)據(jù)表，編寫如下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"
}并發(fā)讀取文件
基本邏輯是主函數(shù)讀取文件夾下面的所有文件，然后用循環(huán)開啟goroutine并傳入文件名和數(shù)據(jù)庫(kù)指針，goroutine中按行讀取每個(gè)文件并將其JSON數(shù)據(jù)轉(zhuǎn)換為結(jié)構(gòu)體，在調(diào)用Gorm寫入Oracle數(shù)據(jù)庫(kù)。這里用Golang的等待組來(lái)同步主函數(shù)與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) // 加載所有文件名
? ? ? ? // 循環(huán)創(chuàng)建goroutine
?? ?for i, v := range files {
?? ??? ?wg.Add(1)
? ? ? ? ? ? ? ? // 將數(shù)據(jù)庫(kù)指針和文件名傳給goroutine處理
?? ??? ?go writeRecord(db, path+v)
?? ?}
?? ?wg.Wait() // 等待所有g(shù)oroutine執(zhí)行完成
?? ?log.Println("over")
}將數(shù)據(jù)寫入數(shù)據(jù)庫(kù)
由于這些文件中可能有重復(fù)的數(shù)據(jù)，所以這里調(diào)用了Gorm的Clauses設(shè)置，當(dāng)有主鍵重復(fù)的數(shù)據(jù)什么都不做，有些情況下主鍵相同但是更新了某些字段，這時(shí)可以用Clauses設(shè)置主鍵重復(fù)時(shí)進(jìn)行更新操作。雖然主鍵重復(fù)時(shí)什么都不做，但是db的執(zhí)行結(jié)果也會(huì)包含"unique constraint"錯(cuò)誤，所以在錯(cuò)誤處理時(shí)要排除主鍵沖突的情況，把其他錯(cuò)誤（如字段太長(zhǎng)或類型不匹配）記錄下來(lái)。
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 // 記錄出錯(cuò)的行數(shù)
for scanner.Scan() {
var shop model.ShopInfo
iter++
// 調(diào)用json.Unmarshal()將文本轉(zhuǎn)換為結(jié)構(gòu)體
if err = json.Unmarshal([]byte(scanner.Text()), &shop); err != nil {
log.Println("轉(zhuǎn)換錯(cuò)誤--->" + scanner.Text())
return
}
// 用clause設(shè)置當(dāng)發(fā)生ID沖突時(shí)什么都不做
res := db.Clauses(clause.OnConflict{DoNothing: true}).Create(&shop)
// 雖然ID相同時(shí)程序不會(huì)停止，但是還是有錯(cuò)誤返回
// 所以這里排除ID沖突錯(cuò)誤，將其他錯(cuò)誤（字段沖突）打印出來(lái)
if res.Error != nil && !strings.Contains(res.Error.Error(), "unique constraint") {
log.Println("插入出錯(cuò)--->" + shop.ShopId + " 在" + filename + "第" + strconv.Itoa(iter) + "行")
return
}
}
}
完整main.go代碼
將上面每一步整合后得到完整的主函數(shù)代碼如下：
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 // 記錄出錯(cuò)的行數(shù)
?? ?for scanner.Scan() {
?? ??? ?var shop model.ShopInfo
?? ??? ?iter++
?? ??? ?// 調(diào)用json.Unmarshal()將文本轉(zhuǎn)換為結(jié)構(gòu)體
?? ??? ?if err = json.Unmarshal([]byte(scanner.Text()), &shop); err != nil {
?? ??? ??? ?log.Println("轉(zhuǎn)換錯(cuò)誤--->" + scanner.Text())
?? ??? ??? ?return
?? ??? ?}
?? ??? ?// 用clause設(shè)置當(dāng)發(fā)生ID沖突時(shí)什么都不做
?? ??? ?res := db.Clauses(clause.OnConflict{DoNothing: true}).Create(&shop)
?? ??? ?// 雖然ID相同時(shí)程序不會(huì)停止，但是還是有錯(cuò)誤返回
?? ??? ?// 所以這里排除ID沖突錯(cuò)誤，將其他錯(cuò)誤（字段沖突）打印出來(lái)
?? ??? ?if res.Error != nil && !strings.Contains(res.Error.Error(), "unique constraint") {
?? ??? ??? ?log.Println("插入出錯(cuò)--->" + shop.ShopId + " 在" + filename + "第" + strconv.Itoa(iter) + "行")
?? ??? ??? ?return
?? ??? ?}
?? ?}
}測(cè)試運(yùn)行
go run ./main.go運(yùn)行過(guò)程非?？?，十幾萬(wàn)條數(shù)據(jù)幾分鐘就寫完了，并且CPU占用率100%，證明非常有效的利用了并發(fā)優(yōu)勢(shì)。若是文件數(shù)量太多（上千個(gè)）的話會(huì)創(chuàng)建非常多goroutine，可能消耗非常多系統(tǒng)資源，可以在循環(huán)創(chuàng)建goroutine時(shí)進(jìn)行限制，只創(chuàng)建30或50個(gè)，一個(gè)goroutine結(jié)束后再給它傳入一個(gè)新的文件名。
到此這篇關(guān)于Golang并發(fā)讀取文件數(shù)據(jù)并寫入數(shù)據(jù)庫(kù)的項(xiàng)目實(shí)踐的文章就介紹到這了,更多相關(guān)Golang并發(fā)讀取并寫入內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
gRPC 是 Google 公司基于 Protobuf 開發(fā)的跨語(yǔ)言的開源 RPC 框架,這篇文章主要介紹了go中的protobuf和grpc使用教程,需要的朋友可以參考下2024-08-08
這篇文章主要介紹了golang coroutine 的等待與死鎖用法詳解，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2021-05-05
Go?語(yǔ)言前綴樹實(shí)現(xiàn)敏感詞檢測(cè)
這篇文章主要為大家介紹了Go語(yǔ)言前綴樹實(shí)現(xiàn)敏感詞檢測(cè)實(shí)例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-08-08
Go?語(yǔ)言開發(fā)環(huán)境搭建過(guò)程
這篇文章主要介紹了Go?語(yǔ)言開發(fā)環(huán)境搭建過(guò)程,本文給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友參考下吧2024-03-03
Go Uber靜態(tài)分析工具NilAway使用初體驗(yàn)
這篇文章主要介紹了Go Uber靜態(tài)分析工具NilAway使用詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2024-01-01
在使用 go 語(yǔ)言開發(fā)過(guò)程中，經(jīng)常需要使用到 json 包來(lái)進(jìn)行 json 和 struct 的互相轉(zhuǎn)換，這篇文章主要介紹了go json轉(zhuǎn)換實(shí)踐中遇到的坑，非常具有實(shí)用價(jià)值，需要的朋友可以參考下2018-12-12
golang常用庫(kù)之pkg/errors包第三方錯(cuò)誤處理包案例詳解
這篇文章主要介紹了golang常用庫(kù)之pkg/errors包第三方錯(cuò)誤處理包,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2022-03-03
Golang仿ps實(shí)現(xiàn)獲取Linux進(jìn)程信息
這篇文章主要為大家學(xué)習(xí)介紹了Golang如何仿ps實(shí)現(xiàn)獲取Linux進(jìn)程信息，文中的示例代碼講解詳細(xì)，感興趣的小伙伴可以跟隨小編一起了解一下2023-07-07
正則表達(dá)式是一種強(qiáng)大的模式匹配工具，能夠在文本中進(jìn)行靈活的搜索和替換操作，本文將介紹?Golang?中的正則表達(dá)式語(yǔ)法，包括常用的匹配符號(hào)、模式修飾符以及示例應(yīng)用，希望對(duì)大家有所幫助2023-05-05
這篇文章主要介紹了go-cqhttp環(huán)境配置,包括go-cqhttp安裝及簡(jiǎn)單介紹，本文結(jié)合示例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2022-09-09