用 Go 語言并發處理 CSV 文件到數據庫

作者：磊豐 2025-01-08 08:39:10

開發前端

通過并發處理，我們能夠大大提升 CSV 文件遷移到數據庫的速度。Go 的 goroutines 和通道非常適合這種類型的任務，可以高效地處理 I/O 密集型的操作。

問題背景

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

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

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

主要思路：

讀取 CSV 文件：使用 encoding/csv 包來解析 CSV 文件。
并發處理數據：將 CSV 文件的數據分批次發送到多個 goroutine 中進行并發處理。
數據庫插入：每個 goroutine 從通道中接收數據并將其插入到數據庫中。
同步控制：使用 sync.WaitGroup 來等待所有 goroutine 完成任務。

假設我們的數據庫是 MySQL，使用 github.com/jinzhu/gorm 作為 ORM 庫來處理數據庫插入。我們會定義一個 Contact 結構體來映射數據庫中的表，并用并發的方式將每一行 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 文件并將每一行數據并發地插入到數據庫中。

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()

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

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

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

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

	// 啟動多個 goroutine 來并發處理 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 // 發送數據到通道
		}(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.")
}

4. 代碼說明

a.初始化數據庫：

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

b.讀取 CSV 文件：

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

c.并發處理數據：

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

d.并發插入數據庫：

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

e.關閉通道與等待：

在所有數據都發送到通道后，使用 wg.Wait() 等待所有 goroutine 完成處理。
關閉通道以確保數據庫插入操作可以順利結束。

5. 性能優化

在這個例子中，我們并發地讀取 CSV 文件并將數據插入數據庫，顯著提高了處理速度。但是，對于大型數據集，還可以做更多的性能優化：

批量插入：可以將多個數據條目批量插入數據庫，而不是每次插入一條記錄。批量插入可以顯著減少數據庫的 I/O 操作，提升性能。
控制并發數：通過 semacphore 或者限制通道緩沖區大小，可以控制并發數，避免數據庫被過多并發請求壓垮。
數據庫連接池：確保數據庫連接池的配置合理，避免過多的并發連接造成數據庫連接耗盡。

6. 總結