Golang并发读取文件数据并写入数据库的项目实践
作者:Monkey_D_Newdun 发布时间:2024-01-29 07:22:50
需求
最近接到一个任务,要把一批文件中的十几万条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
猜你喜欢
- 我页面上有控制了只能输入数字的控件,禁止了输入法切换的,但是搜狗的云输入却控制不了,有没有办法在页面里面禁止它运行啊?发现这玩意儿真的很讨厌
- 1、跳过迭代对象的开头string_from_file = """ // Wooden: ...
- ini文件是windows中经常使用的配置文件,主要的格式为:[Section1]option1 : value1option2 : val
- 前言本文主要给大家介绍了Go语言中函数new与make的使用和区别,关于Go语言中new和make是内建的两个函数,主要用来创建分配类型内存
- 这篇文章主要介绍了python使用rsa非对称加密过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要
- 初学 Python 的开发者经常会发现很多 Python 函数中用到了 yield 关键字,然而,带有 yield 的函数执行流程却和普通函
- 如下所示:# coding=utf-8import pandas as pd# 读取csv文件 3列取名为 name,sex,births,
- 1.字符串大小写转换string.title() #将字符串中所有单词的首字母以大写形式显示string.upper() #将字符串中所有字
- 英文版:File -> settings -> Editor -> File Encodings首先打开设置:文件 -&g
- 本文实例讲述了python使用any判断一个对象是否为空的方法。分享给大家供大家参考。具体实现代码如下:>>> eth =
- Python常用的数据结构,有如下几种。但是我们用的最多的,还是字符串、列表、字典这3种。其实学习任何一门编程语言,最基础的就是学习它的数据
- Stochastic Depth论文:Deep Networks with Stochastic Depth本文的正则化针对于ResNet中
- 简述:Django的admin可以提供一个强大的后台管理功能,可以在web界面对数据库进行操作,我们需要修改admin.py将要操作的数据表
- 先给大家介绍下sqlserver给表添加新字段、给表和字段添加备注、更新备注及查询备注,代码如下所示:-- 添加新字段及字段备注的语法USE
- 目录1. 现实背景2. 约定3. 配置 master3.1. 配置启动参数3.2. 重启服务使参数生效3.3. 以 root 身份登录,创建
- 1 读/写文件NumPy文件读写主要有二进制的文件读写和文件列表形式的数据读写两种形式1、二进制的文件读写save函数是以二进制的格式保存数
- 在做项目的过程中,我们经常会建立各种各样的规范,以方便团队之间更好的合作更好的完成项目;同样我们也经常会听到各种各样的协议,比如Google
- 实训课期间忙里偷闲的学习了python的selenium包,唯一一点不好是要自己去查英文文档,明摆着欺负我这种英语不好的,想着用谷歌翻译一下
- preface:做着最近的任务,对数据处理,做些简单的提特征,用机器学习算法跑下程序得出结果,看看哪些特征的组合较好,这一系列流程必然要用到
- 问题描述如下:解决方案如下:下图中字体调整为18及以上效果如下:来源:https://blog.csdn.net/appleyuchi/ar