Bagaimana untuk menangani pengurusan kumpulan sambungan sambungan pangkalan data serentak dalam bahasa Go?

Bahasa Go ialah bahasa pengaturcaraan serentak yang cekap dengan ciri seperti benang ringan (goroutine) dan saluran, yang sangat sesuai untuk menangani isu konkurensi. Dalam pembangunan sebenar, pengurusan sambungan pangkalan data adalah isu utama. Pengumpulan sambungan ialah penyelesaian biasa yang boleh meningkatkan penggunaan semula dan prestasi sambungan pangkalan data. Artikel ini akan memperkenalkan cara menggunakan pengumpulan sambungan untuk mengurus sambungan pangkalan data serentak dalam bahasa Go dan memberikan contoh kod khusus.

Idea reka bentuk kumpulan sambungan

Matlamat teras kumpulan sambungan adalah untuk merealisasikan penggunaan semula sambungan dan mengelak daripada membuat dan menutup sambungan pangkalan data dengan kerap. Dalam pengaturcaraan serentak, setiap goroutine boleh memohon dan mengembalikan sambungan pangkalan data secara bebas, jadi kumpulan sambungan perlu mempunyai fungsi berikut:

1. Mulakan kumpulan sambungan: Apabila program bermula, cipta nombor tertentu sambungan pangkalan data terlebih dahulu dan letakkan mereka ke tengah kolam sambungan.
2. Pengembangan dan pengecutan dinamik: Mengikut keperluan sebenar, kumpulan sambungan boleh meningkatkan atau mengurangkan sambungan pangkalan data yang tersedia secara dinamik.
3. Gunakan dan kembalikan sambungan: Setiap goroutine boleh mengeluarkan sambungan dari kolam sambungan dan mengembalikan sambungan ke kolam sambungan selepas digunakan.

Pelaksanaan kumpulan sambungan

Pertama, kita perlu menentukan struktur untuk mewakili kumpulan sambungan. Struktur mengandungi medan berikut:

• kod: Barisan gilir sambungan dalam kumpulan sambungan, dilaksanakan menggunakan saluran. pool：连接池中的连接队列，使用通道来实现。
• capacity：连接池中连接的最大容量。
• count：当前连接池中的连接数。

type Pool struct {
    pool     chan *sql.DB
    capacity int
    count    int
}

接下来，我们可以定义一些连接池需要的方法：

• NewPool：初始化连接池，创建并放入指定数量的数据库连接。
• Get：从连接池中获取一个数据库连接。
• Put：将一个数据库连接放回连接池中。
• Expand：动态增加连接池中的连接容量。
• Shrink
kapasiti: Kapasiti maksimum sambungan dalam kumpulan sambungan.

kira: Bilangan sambungan dalam kumpulan sambungan semasa.

func NewPool(dbURL string, capacity int) (*Pool, error) {
    // 创建连接池并初始化
    pool := make(chan *sql.DB, capacity)
    for i := 0; i < capacity; i++ {
        db, err := sql.Open("mysql", dbURL)
        if err != nil {
            return nil, err
        }
        pool <- db
    }

    return &Pool{
        pool:     pool,
        capacity: capacity,
        count:    capacity,
    }, nil
}

func (p *Pool) Get() (*sql.DB, error) {
    // 从连接池获取一个连接
    db := <-p.pool
    p.count--

    return db, nil
}

func (p *Pool) Put(db *sql.DB) {
    // 将连接放回连接池
    p.pool <- db
    p.count++
}

func (p *Pool) Expand() error {
    // 增加连接池中的连接容量
    db, err := sql.Open("mysql", dbURL)
    if err != nil {
        return err
    }
    p.pool <- db
    p.count++

    return nil
}

func (p *Pool) Shrink() error {
    // 减少连接池中的连接容量
    db := <-p.pool
    db.Close()
    p.count--

    return nil
}

NewPool: Mulakan kumpulan sambungan, buat dan masukkan bilangan sambungan pangkalan data yang ditentukan.

Dapatkan: Dapatkan sambungan pangkalan data daripada kumpulan sambungan.

Letak: Letakkan semula sambungan pangkalan data ke dalam kumpulan sambungan.

func main() {
    dbURL := "username:password@tcp(hostname:port)/dbname"
    capacity := 10

    // 创建连接池
    pool, err := NewPool(dbURL, capacity)
    if err != nil {
        log.Fatal(err)
    }

    // 并发查询
    var wg sync.WaitGroup
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()

            // 从连接池获取一个连接
            db, err := pool.Get()
            if err != nil {
                log.Println(err)
                return
            }
            defer pool.Put(db)

            // 执行查询
            rows, err := db.Query("SELECT * FROM users")
            if err != nil {
                log.Println(err)
                return
            }
            defer rows.Close()

            // 处理查询结果
            for rows.Next() {
                var name string
                err := rows.Scan(&name)
                if err != nil {
                    log.Println(err)
                    return
                }
                log.Println("Query result:", name)
            }
        }(i)
    }

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