Go学习笔记 - 并发查询单表

2021-12-02

全文 ≈ 512字 | 预计阅读时长 2分钟

最近接触一个项目，需要每天定时把PgSql一张数据量超大的数据表同步到ClickHouse 思考这种场景下怎么提高同步速度。

按主键分页

思路

按照主键排序分区间查询

代码

package main

import (
	"fmt"
	"math"
	"strconv"
	"sync"
)

func main() {
	maxId := 203202
	pageNum := 1500
	num, _ := strconv.ParseFloat(fmt.Sprintf("%.2f", float64(maxId)/float64(pageNum)), 64)
	maxPage := int(math.Ceil(num))

	wg := &sync.WaitGroup{}
	for i := 0; i < maxPage; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()

			startId := (i * pageNum)
			endId := startId + pageNum
			if endId > maxId {
				endId = maxId
			}
			sql := fmt.Sprintf("select * from logs where id >= %d and id <= %d", startId, endId)

			fmt.Printf("第 %d 页 SQL: %s\n", i+1, sql)
		}(i)
	}
	wg.Wait()
}

优点

命中索引查询
缺点

PgSql设置主键会影响插入和查询速度，这也是为什么PgSql数据表一般不设置主键的原因

游标

思路

多个协程获取同一个游标的数据

代码

package main

import (
	"fmt"
	"project/postgresql"
	"math"
	"strconv"
	"sync"
)

func main() {
	// pgSql client
	cli := postgresql.GetClient()

	// trx
	tx, err := cli.Begin()
	if err != nil {
		panic(err)
	}

	// 计算多少个coroutine
	cnt := 0
	cli.QueryRow("select count(*) from logs").Scan(&cnt)
	pageNum := 1500
	num, _ := strconv.ParseFloat(fmt.Sprintf("%.2f", float64(cnt)/float64(pageNum)), 64)
	maxPage := int(math.Ceil(num))

	// cursor
	cursorName := "cursor_name"
	cursorQuery := "select * from logs"
	tx.Query("DECLARE " + cursorName + " CURSOR FOR " + cursorQuery)

	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	for i := 1; i <= maxPage; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()

			// 互斥锁
			lock.Lock()

			// 每次取{pageNum}条
			rows, err := tx.Query("FETCH FORWARD " + strconv.Itoa(pageNum) + " FROM " + cursorName)
			if err != nil {
				panic(err)
			}
			defer rows.Close()

			// 处理迭代逻辑
			// for rows.Next() {

			// }
			fmt.Printf("第 %d 页 结果: %v\n", i, rows)

			// 解锁
			// TODO 一定要迭代完成后再解锁
			lock.Unlock()
		}(i)
	}
	wg.Wait()
	tx.Commit()
}

优点

能保证数据的完整，可靠性较高
缺点

必须等待结果集迭代完成后才能处理下一个

为啥不用 offset, limit

因为offset, limit会扫全表 233