使用Go語言實現并發(fā)處理CSV文件到數據庫

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

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

問題背景

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

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

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

主要思路

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

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

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

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

的方式將每一行 CSV 數據插入到數據庫。

示例代碼

1. 安裝必要的依賴

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

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

2. 數據庫模型定義

我們先定義一個 Contact 結構體，它會對應數據庫中的聯系人表。

package main

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

// Contact 是數據庫中表的模型
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 數據庫
	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 文件并將每一行數據并發(fā)地插入到數據庫中。

package main

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

// 處理 CSV 文件并將數據插入數據庫
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ā)送每行數據
	ch := make(chan Contact, len(records))

	// 啟動多個 goroutine 來并發(fā)處理 CSV 數據
	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ā)送數據到通道
		}(records[i])
	}

	// 啟動一個 goroutine 來將通道中的數據插入到數據庫
	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() {
	// 初始化數據庫
	db, err := initDB()
	if err != nil {
		fmt.Println("Failed to connect to database:", err)
		return
	}
	defer db.Close()

	// 處理 CSV 文件并將數據遷移到數據庫
	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.")
}

代碼說明

初始化數據庫：

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

讀取 CSV 文件：

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

并發(fā)處理數據：

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

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

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

關閉通道與等待：