You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用GoLang迁移TimeScaleDB速度过慢，求优化方案

avatar
开发者阿泽
2026-7-12
问题描述

需要将包含150万行数据的表从旧TimeScaleDB实例迁移至新实例。当前实现逻辑为：从offset=0开始，每次查询200行加载到内存后执行插入，经统计每200行的查询-插入循环耗时约1分3秒至1分7秒。调试发现当前代码对每条查询到的数据逐行执行插入操作，且本人为GoLang新手，代码可能存在不规范之处。

.env配置文件 

FROM_TIMESCALE_HOST=
FROM_TIMESCALE_PORT=
FROM_TIMESCALE_DATABASE=
FROM_TIMESCALE_USERNAME=
FROM_TIMESCALE_PASSWORD=
MIGRATED_TABLE=table
TO_TIMESCALE_HOST=
TO_TIMESCALE_PORT=
TO_TIMESCALE_DATABASE=
TO_TIMESCALE_USERNAME=
TO_TIMESCALE_PASSWORD=
LIMITED_QUANTITY_ROW=

迁移代码（go-migration.go） 

package main

import (
    "database/sql"
    "fmt"
    "log"
    "os"
    "strconv"
    "time"

    "github.com/joho/godotenv"
    _ "github.com/lib/pq"
)

type statistics struct {
    TimestampTZ  time.Time `db:"time" json:"time"`
    ContentID    int64     `db:"content_id" json:"content_id"`
    LikeCount    int64     `db:"like_count" json:"like_count"`
    CommentCount int64     `db:"comment_count" json:"comment_count"`
    ShareCount   int64     `db:"share_count" json:"share_count"`
    ViewCount    int64     `db:"view_count" json:"view_count"`
    SaveCount    int64     `db:"save_count" json:"save_count"`
}

// type statistics struct {
//  TimestampTZ  string
//  ContentID    int64
//  LikeCount    sql.NullInt64
//  CommentCount sql.NullInt64
//  ShareCount   sql.NullInt64
//  ViewCount    sql.NullInt64
//  SaveCount    sql.NullInt64
// }

func main() {
    // Load environment variables from .env file
    From_DB, To_DB, migrated_table, limited_quantity_row, err := loadEnv()
    if err != nil {
        log.Fatal(err)
    }
    connectionFrom_DB, err := openConnection(From_DB)
    if err != nil {
        log.Fatal(err)
    }
    connectionTo_DB, err := openConnection(To_DB)
    if err != nil {
        log.Fatal(err)
    }

    startTime := time.Now()

    doMigration(connectionFrom_DB, connectionTo_DB, To_DB["TIMESCALE_HOST"], migrated_table, limited_quantity_row, 0)

    endTime := time.Now()

    fmt.Println(endTime.Sub(startTime))
}

func loadEnv() (map[string]string, map[string]string, string, int, error) {
    err := godotenv.Load()
    if err != nil {
        return nil, nil, "", 0, fmt.Errorf("error loading .env file: %s", err)
    }
    fmt.Printf("Loaded .env file!\n\n")

    migratedTable, ok := os.LookupEnv("MIGRATED_TABLE")
    if !ok {
        return nil, nil, "", 0, fmt.Errorf("missing migrated table value in .env")
    }

    limitedQuantityQueriedRowstr, ok := os.LookupEnv("LIMITED_QUANTITY_ROW")
    if !ok {
        return nil, nil, "", 0, fmt.Errorf("missing limited quantity row value in .env")
    }
    limitedQuantityRow, err := strconv.Atoi(limitedQuantityQueriedRowstr)
    if err != nil {
        return nil, nil, "", 0, fmt.Errorf("error converting limited quantity row to integer: %s", err)
    }

    fromDB, err := getDBInformation("FROM")
    if err != nil {
        return nil, nil, "", 0, fmt.Errorf("error loading From DB information: %s", err)
    }

    toDB, err := getDBInformation("TO")
    if err != nil {
        return nil, nil, "", 0, fmt.Errorf("error loading To DB information: %s", err)
    }

    return fromDB, toDB, migratedTable, limitedQuantityRow, nil
}
func getDBInformation(input string) (map[string]string, error) {
    env := make(map[string]string)

    if TimescaleHost, ok := os.LookupEnv(input + "_TIMESCALE_HOST"); !ok {
        return nil, fmt.Errorf("missing host ")
    } else {
        env["TIMESCALE_HOST"] = TimescaleHost
    }

    if TimescalePort, ok := os.LookupEnv(input + "_TIMESCALE_PORT"); !ok {
        return nil, fmt.Errorf("missing port ")
    } else {
        env["TIMESCALE_PORT"] = TimescalePort
    }

    if TimescaleDatabase, ok := os.LookupEnv(input + "_TIMESCALE_DATABASE"); !ok {
        return nil, fmt.Errorf("missing database's name ")
    } else {
        env["TIMESCALE_DATABASE"] = TimescaleDatabase
    }

    if TimescaleUsername, ok := os.LookupEnv(input + "_TIMESCALE_USERNAME"); !ok {
        return nil, fmt.Errorf("missing username ")
    } else {
        env["TIMESCALE_USERNAME"] = TimescaleUsername
    }

    if TimescalePassword, ok := os.LookupEnv(input + "_TIMESCALE_PASSWORD"); !ok {
        return nil, fmt.Errorf("missing password ")
    } else {
        env["TIMESCALE_PASSWORD"] = TimescalePassword
    }
    // fmt.Println(env)
    fmt.Println(input + " TIMESCALE Information loaded! \n")
    return env, nil

}

func openConnection(env map[string]string) (*sql.DB, error) {
    // Construct the database connection string
    // connStr := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=require", env["TIMESCALE_HOST"], env["TIMESCALE_PORT"], env["TIMESCALE_USERNAME"], env["TIMESCALE_PASSWORD"], env["TIMESCALE_DATABASE"])
    connStr := fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=disable", env["TIMESCALE_USERNAME"], env["TIMESCALE_PASSWORD"], env["TIMESCALE_HOST"], env["TIMESCALE_PORT"], env["TIMESCALE_DATABASE"])
    // fmt.Println(connStr)
    // Open the database connection
    db, err := sql.Open("postgres", connStr)
    if err != nil {
        db.Close()
        fmt.Printf("Can't connect to the database at %s.\n", env["TIMESCALE_HOST"])
        return nil, err
    }

    // // Check the database connection
    // err = db.Ping()
    // if err != nil {
    //  db.Close()
    //  fmt.Printf("Can't ping to the database at %s\n", env["TIMESCALE_HOST"])
    //  return nil, err
    // }
    fmt.Printf("Connected to the %s database !\n\n", env["TIMESCALE_HOST"])
    return db, nil
}

func doMigration(connectionFrom_DB *sql.DB, connectionTo_DB *sql.DB, host string, migrated_table string, limited_quantity_row int, start_offset int64) {
    fmt.Printf("Starting query data from %s\n", migrated_table)
    //Prepare the insert statement
    insertStatement := fmt.Sprintf("INSERT INTO %s (time, content_id, like_count, comment_count, share_count, view_count, save_count) VALUES ($1,$2,$3,$4,$5,$6,$7)", migrated_table)
    // insertStatement :=fmt.Sprintf("INSERT INTO %s (time, content_id, like_count, comment_count, share_count, view_count, save_count) VALUES (?,?,?,?,?,?,?))", migrated_table)
    // fmt.Println(insertStatement)

    // Query with limited_quantity_row start at start_offset and ascending start_offset after each loop
    for {
        if start_offset >= 59 {
            // Just stop to check
            break
        }
        //===================================================================
        // do the SELECT queey to the db
        statisticsList, error := queryBD1(connectionFrom_DB, migrated_table, limited_quantity_row, start_offset)
        if error != nil {
            log.Fatal(error)
        }
        //===================================================================
        if len(statisticsList) == 0 {
            // No more rows, break the loop
            break
        }
        //===================================================================
        // Insert the data into the destination DB
        queryDB2(connectionTo_DB, insertStatement, statisticsList)
        //===================================================================
        // Increment the start_offset for the next iteration
        start_offset += int64(limited_quantity_row)
    }
}

func printLogMigration(stat statistics, index int64) {
    fmt.Printf("%-10d", index)
    fmt.Printf("%20s|%10d|%10d|%10d|%10d|%10d|%10d\n", stat.TimestampTZ.String(), stat.ContentID, stat.LikeCount, stat.CommentCount, stat.ShareCount, stat.ViewCount, stat.SaveCount)
}
func queryBD1(connection *sql.DB, migrated_table string, limited_quantity_row int, start_offset int64) ([]statistics, error) {
    startTime := time.Now()
    rows, queryError := connection.Query(fmt.Sprintf("SELECT * FROM %s LIMIT %d OFFSET %d", migrated_table, limited_quantity_row, start_offset))
    if queryError != nil {
        log.Fatal(queryError)
    }

    // Prepare the statement for inserting into the destination DB
    index := start_offset + 1
    var statsList []statistics
    for rows.Next() {
        var stats statistics
        err := rows.Scan(
            &stats.TimestampTZ,
            &stats.ContentID,
            &stats.LikeCount,
            &stats.CommentCount,
            &stats.ShareCount,
            &stats.ViewCount,
            &stats.SaveCount,
        )
        if err != nil {
            log.Fatal(err)
        }

        printLogMigration(stats, index)
        index++
        // stmt.ExecContext()
        statsList = append(statsList, stats)
    }
    endTime := time.Now()
    fmt.Printf("time to query %f \n", endTime.Sub(startTime).Seconds())
    return statsList, nil
}
func queryDB2(connection *sql.DB, insertStatement string, statisticsList []statistics) {
    
    if err != nil {
        log.Fatal(err)
    }
    defer stmt.Close()
    for _, stats := range statisticsList {
        _, execErr := stmt.Exec(
            stats.TimestampTZ,
            stats.ContentID,
            stats.LikeCount,
            stats.CommentCount,
            stats.ShareCount,
            stats.ViewCount,
            stats.SaveCount,
        )
        if execErr != nil {
            log.Fatal(execErr)
        }
    }
}
优化思路

1. 批量插入替代逐行插入

当前queryDB2函数逐行执行INSERT，会产生大量数据库交互开销。改为批量插入，将整批次数据合并为一条INSERT语句：

import "strings"

func queryDB2(connection *sql.DB, insertStatement string, statisticsList []statistics) {
    if len(statisticsList) == 0 {
        return
    }
    // 构造批量插入的占位符和参数
    valuesPlaceholders := make([]string, 0, len(statisticsList))
    args := make([]interface{}, 0, len(statisticsList)*7)
    for i, stats := range statisticsList {
        idx := i*7 + 1
        valuesPlaceholders = append(valuesPlaceholders, fmt.Sprintf("($%d,$%d,$%d,$%d,$%d,$%d,$%d)", idx, idx+1, idx+2, idx+3, idx+4, idx+5, idx+6))
        args = append(args,
            stats.TimestampTZ,
            stats.ContentID,
            stats.LikeCount,
            stats.CommentCount,
            stats.ShareCount,
            stats.ViewCount,
            stats.SaveCount,
        )
    }
    batchInsertStmt := fmt.Sprintf("%s VALUES %s", insertStatement, strings.Join(valuesPlaceholders, ","))
    _, err := connection.Exec(batchInsertStmt, args...)
    if err != nil {
        log.Fatal(err)
    }
}

批量插入能大幅减少数据库往返次数，直接提升插入效率。

2. 替换OFFSET分页为游标分页

当前使用LIMIT ... OFFSET ...分页，当偏移量接近150万行时，数据库需要扫描大量无关数据才能定位目标行，查询效率会急剧下降。利用表中有序字段（比如time）实现游标分页：

func doMigration(connectionFrom_DB *sql.DB, connectionTo_DB *sql.DB, host string, migrated_table string, limited_quantity_row int, _ int64) {
    fmt.Printf("Starting query data from %s\n", migrated_table)
    insertStatement := fmt.Sprintf("INSERT INTO %s (time, content_id, like_count, comment_count, share_count, view_count, save_count) VALUES ($1,$2,$3,$4,$5,$6,$7)", migrated_table)

    lastTime := time.Time{}
    for {
        var rows *sql.Rows
        var err error
        // 根据游标位置构建查询语句
        if lastTime.IsZero() {
            rows, err = connectionFrom_DB.Query(fmt.Sprintf("SELECT * FROM %s ORDER BY time LIMIT %d", migrated_table, limited_quantity_row))
        } else {
            rows, err = connectionFrom_DB.Query(fmt.Sprintf("SELECT * FROM %s WHERE time > $1 ORDER BY time LIMIT %d", migrated_table, limited_quantity_row), lastTime)
        }
        if err != nil {
            log.Fatal(err)
        }

        var statsList []statistics
        index := int64(0)
        for rows.Next() {
            var stats statistics
            err := rows.Scan(
                &stats.TimestampTZ,
                &stats.ContentID,
                &stats.LikeCount,
                &stats.CommentCount,
                &stats.ShareCount,
                &stats.ViewCount,
                &stats.SaveCount,
            )
            if err != nil {
                log.Fatal(err)
            }
            statsList = append(statsList, stats)
            index++
        }
        rows.Close()

        if len(statsList) == 0 {
            break
        }
        queryDB2(connectionTo_DB, insertStatement, statsList)
        // 更新游标为当前批次最后一行的时间
        lastTime = statsList[len(statsList)-1].TimestampTZ
    }
}

这种方式每次查询仅扫描必要数据，查询性能更稳定。

3. 调整批次大小

当前每次查询200行，可以尝试增大批次（比如1000行或5000行），减少循环次数和数据库交互次数。需注意内存占用，避免加载过多数据导致内存溢出。

4. 优化数据库连接池

Go的sql.DB是连接池，默认配置可能无法发挥最优性能，可调整参数：

func openConnection(env map[string]string) (*sql.DB, error) {
    connStr := fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=disable", env["TIMESCALE_USERNAME"], env["TIMESCALE_PASSWORD"], env["TIMESCALE_HOST"], env["TIMESCALE_PORT"], env["TIMESCALE_DATABASE"])
    db, err := sql.Open("postgres", connStr)
    if err != nil {
        db.Close()
        fmt.Printf("Can't connect to the database at %s.\n", env["TIMESCALE_HOST"])
        return nil, err
    }
    // 调整连接池参数
    db.SetMaxOpenConns(20)   // 最大打开连接数，根据数据库承载能力调整
    db.SetMaxIdleConns(10)   // 最大空闲连接数
    db.SetConnMaxLifetime(30 * time.Minute) // 连接生命周期
    // 验证连接
    if err := db.Ping(); err != nil {
        db.Close()
        fmt.Printf("Can't ping to the database at %s\n", env["TIMESCALE_HOST"])
        return nil, err
    }
    fmt.Printf("Connected to the %s database !\n\n", env["TIMESCALE_HOST"])
    return db, nil
}

5. 关闭不必要的日志输出

当前printLogMigration函数逐行打印数据，会消耗大量IO资源拖慢迁移速度。迁移时可关闭该日志，或改为每隔N批次打印一次进度。

6. 用事务包裹批量插入

将每批次插入操作包裹在事务中，减少事务提交开销，同时保证数据一致性：

func queryDB2(connection *sql.DB, insertStatement string, statisticsList []statistics) {
    if len(statisticsList) == 0 {
        return
    }
    tx, err := connection.Begin()
    if err != nil {
        log.Fatal(err)
    }
    defer tx.Rollback()

    valuesPlaceholders := make([]string, 0, len(statisticsList))
    args := make([]interface{}, 0, len(statisticsList)*7
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
火山引擎 最新活动
全面开放
Doubao-Seedance-2.5全新模型升级

模型再升级,30秒超长叙事, 模态参考扩容

HOT
方舟Codingplan

模型自由，工具不限，最新支持 Deepseek-V4 系列、GLM-5.3 系列

HOT
方舟Agent plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列

HOT
新用户特惠专场

大模型19元起，Al应用9.9元畅享，新人首购爆款尽享优惠

最近更新时间：2026.07.12 10:25:55