本篇内容介绍了“怎么使用Golang并发读取文件数据并写入数据库”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
项目结构
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 modeltype 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.WaitGroupfunc 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 mainimport ( "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.WaitGroupfunc 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并发读取文件数据并写入数据库”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注编程网网站，小编将为大家输出更多高质量的实用文章！
