網站首頁 編程語言 正文
需求
最近接到一個任務，要把一批文件中的十幾萬條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, "\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
}
}
}
完整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, "\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結束后再給它傳入一個新的文件名。
原文鏈接：https://blog.csdn.net/Monkey_D_Newdun/article/details/122677101
相關推薦
- 2022-04-06 .NET?Core使用CZGL.SystemInfo庫獲取主機運行資源_基礎應用
- 2022-08-15 springboot切換為redis緩存
- 2023-04-19 Git綁定遠程倉庫報錯error: remote origin already exists.
- 2022-07-19 react組件通訊的三種方式props:父組件和子組件互相通訊、兄弟組件通訊
- 2023-07-14 element組件中的時間選擇器，禁用選擇時間，picker-options屬性詳解，時間選擇器范圍
- 2022-05-05 RabbitMQ的Web管理與監控簡介_web2.0
- 2022-05-11 pg分區表的實踐
- 2022-07-10 Popconfirm氣泡確認框無法觸發confirm函數
- 最近更新
-
- window11 系統安裝 yarn
- 超詳細win安裝深度學習環境2025年最新版（
- Linux 中運行的top命令 怎么退出？
- MySQL 中decimal 的用法？ 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎操作-- 運算符，流程控制 Flo
- 1. Int 和Integer 的區別，Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權
- redisson分布式鎖中waittime的設
- maven:解決release錯誤：Artif
- restTemplate使用總結
- Spring Security之安全異常處理
- MybatisPlus優雅實現加密？
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務發現-Nac
- Spring Security之基于HttpR
- Redis 底層數據結構-簡單動態字符串（SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應用詳解
- 聊聊消息隊列，發送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支
提供CDN加速