使用Go語言實現(xiàn)并發(fā)處理CSV文件到數(shù)據(jù)庫

更新時間：2025年01月20日 10:31:37 作者：Ai?編碼

Go?語言的?goroutine?和通道（channel）非常適合用來并發(fā)地處理數(shù)據(jù),本文將通過簡單示例介紹一下如何使用Go語言并發(fā)地處理?CSV?文件并將數(shù)據(jù)插入到數(shù)據(jù)庫中,感興趣的可以了解下

問題背景

假設你擁有一個包含大量聯(lián)系人信息的 CSV 文件，需要將這些信息遷移到數(shù)據(jù)庫中。這些聯(lián)系人信息可能包含姓名、電話號碼、郵箱地址等。如果使用傳統(tǒng)的單線程方式，逐條處理數(shù)據(jù)，遷移過程可能會非常緩慢，尤其是在數(shù)據(jù)量很大時。

在處理大量的 CSV 文件數(shù)據(jù)并遷移到數(shù)據(jù)庫時，使用并發(fā)可以顯著提升處理效率。Go 語言的 goroutine 和通道（channel）非常適合用來并發(fā)地處理數(shù)據(jù)。

下面我將給出一個示例，展示如何使用 Go 語言并發(fā)地處理 CSV 文件，并將數(shù)據(jù)插入到數(shù)據(jù)庫中。

主要思路

讀取 CSV 文件：使用 encoding/csv 包來解析 CSV 文件。

并發(fā)處理數(shù)據(jù)：將 CSV 文件的數(shù)據(jù)分批次發(fā)送到多個 goroutine 中進行并發(fā)處理。

數(shù)據(jù)庫插入：每個 goroutine 從通道中接收數(shù)據(jù)并將其插入到數(shù)據(jù)庫中。

同步控制：使用 sync.WaitGroup 來等待所有 goroutine 完成任務。
假設我們的數(shù)據(jù)庫是 MySQL，使用 github.com/jinzhu/gorm 作為 ORM 庫來處理數(shù)據(jù)庫插入。我們會定義一個 Contact 結構體來映射數(shù)據(jù)庫中的表，并用并發(fā)

的方式將每一行 CSV 數(shù)據(jù)插入到數(shù)據(jù)庫。

示例代碼

1. 安裝必要的依賴

首先，你需要安裝 gorm 和 csv 相關的包：

go get github.com/jinzhu/gorm
go get github.com/jinzhu/gorm/dialects/mysql
go get encoding/csv

2. 數(shù)據(jù)庫模型定義

我們先定義一個 Contact 結構體，它會對應數(shù)據(jù)庫中的聯(lián)系人表。

package main

import (
	"github.com/jinzhu/gorm"
	_ "github.com/jinzhu/gorm/dialects/mysql"
	"fmt"
)

// Contact 是數(shù)據(jù)庫中表的模型
type Contact struct {
	ID        uint   `gorm:"primary_key"`
	Name      string `gorm:"size:255"`
	Phone     string `gorm:"size:255"`
	Email     string `gorm:"size:255"`
}

func initDB() (*gorm.DB, error) {
	// 使用 MySQL 數(shù)據(jù)庫
	db, err := gorm.Open("mysql", "user:password@/dbname?charset=utf8&parseTime=True&loc=Local")
	if err != nil {
		return nil, err
	}

	// 自動遷移表結構
	db.AutoMigrate(&Contact{})
	return db, nil
}

3. 讀取 CSV 文件并處理

接下來，我們需要讀取 CSV 文件并將每一行數(shù)據(jù)并發(fā)地插入到數(shù)據(jù)庫中。

package main

import (
	"encoding/csv"
	"fmt"
	"os"
	"strings"
	"sync"
)

// 處理 CSV 文件并將數(shù)據(jù)插入數(shù)據(jù)庫
func processCSV(filePath string, db *gorm.DB) error {
	// 打開 CSV 文件
	file, err := os.Open(filePath)
	if err != nil {
		return err
	}
	defer file.Close()

	// 創(chuàng)建 CSV 閱讀器
	reader := csv.NewReader(file)

	// 讀取所有行
	records, err := reader.ReadAll()
	if err != nil {
		return err
	}

	// 使用 WaitGroup 來同步所有的 goroutine
	var wg sync.WaitGroup

	// 通道用于發(fā)送每行數(shù)據(jù)
	ch := make(chan Contact, len(records))

	// 啟動多個 goroutine 來并發(fā)處理 CSV 數(shù)據(jù)
	for i := 1; i < len(records); i++ { // 從 1 開始，跳過標題行
		wg.Add(1)
		go func(record []string) {
			defer wg.Done()
			// 將 CSV 行轉換為 Contact 實例
			contact := Contact{
				Name:  record[0],
				Phone: record[1],
				Email: record[2],
			}
			ch <- contact // 發(fā)送數(shù)據(jù)到通道
		}(records[i])
	}

	// 啟動一個 goroutine 來將通道中的數(shù)據(jù)插入到數(shù)據(jù)庫
	go func() {
		for contact := range ch {
			if err := db.Create(&contact).Error; err != nil {
				fmt.Println("Error inserting record:", err)
			}
		}
	}()

	// 等待所有 goroutine 完成
	wg.Wait()

	// 關閉通道
	close(ch)

	return nil
}

func main() {
	// 初始化數(shù)據(jù)庫
	db, err := initDB()
	if err != nil {
		fmt.Println("Failed to connect to database:", err)
		return
	}
	defer db.Close()

	// 處理 CSV 文件并將數(shù)據(jù)遷移到數(shù)據(jù)庫
	err = processCSV("contacts.csv", db)
	if err != nil {
		fmt.Println("Error processing CSV file:", err)
		return
	}

	fmt.Println("CSV data successfully migrated to the database.")
}

代碼說明

初始化數(shù)據(jù)庫：

initDB 函數(shù)用于初始化 MySQL 數(shù)據(jù)庫連接并進行自動遷移。
我們使用 gorm 來處理數(shù)據(jù)庫操作，模型 Contact 映射到數(shù)據(jù)庫中的 contacts 表。

讀取 CSV 文件：

processCSV 函數(shù)打開并讀取 CSV 文件。然后，它讀取所有的記錄，并將每條記錄通過 goroutine 異步發(fā)送到通道中。
每個 goroutine 都會將一條記錄從 CSV 轉換為 Contact 對象，并將其發(fā)送到通道。

并發(fā)處理數(shù)據(jù)：

sync.WaitGroup 被用來確保所有的 goroutine 完成任務。wg.Add(1) 在啟動每個 goroutine 時調用，wg.Done() 在每個 goroutine 完成時調用。
使用 chan Contact 通道來將數(shù)據(jù)從多個 goroutine 傳遞到數(shù)據(jù)庫插入部分。一個單獨的 goroutine 從通道中接收數(shù)據(jù)并將其插入到數(shù)據(jù)庫。

并發(fā)插入數(shù)據(jù)庫：

每個 goroutine 向通道發(fā)送數(shù)據(jù)，然后另一個 goroutine 從通道中讀取數(shù)據(jù)并將其插入數(shù)據(jù)庫。通過這種方式，多個數(shù)據(jù)庫插入操作是并發(fā)進行的。

關閉通道與等待：