Goアプリケーションは一般的に、同時に作業を処理するためにGoroutineを使用します。
go-mssqldbドライバーとdatabase/sqlパッケージは同時使用を想定して設計されていますが、接続漏れ、データ競合、プール不足を避けるために特定のパターンに従う必要があります。 この記事では、Goroutineの安全性、従業員プール、そしてスムーズなシャットダウンパターンについて解説します。
Goroutineの安全性
sql.DB は並行して使用しても安全です
*sql.DBインスタンスは複数のゴルーチンから同時に使用しても安全です。 内部接続プールを管理し、同期を処理します:
// CORRECT: Share a single *sql.DB across all goroutines.
var db *sql.DB
func main() {
var err error
db, err = sql.Open("sqlserver", connString)
if err != nil {
log.Fatal(err)
}
defer db.Close()
http.HandleFunc("/employees", listEmployees) // Each request runs in its own goroutine.
log.Fatal(http.ListenAndServe(":8080", nil))
}
Warning
リクエストごとやゴルーチンごとに新しい sql.Open を作成しないでください。 各 sql.Open 通話は別個の接続プールを作成します。 リクエストごとにプールを作成するとリソースが無駄になり、サーバー側の接続制限をすぐに使い果たす可能性があります。
sql.Rows、sql.Tx、sql.Conn は並行使用に対して安全ではありません
これらの型は単一の接続を表し、一度に1つのゴルーチンから使用する必要があります:
// WRONG: Sharing rows across goroutines causes data races.
rows, _ := db.QueryContext(ctx, "SELECT BusinessEntityID, FirstName + ' ' + LastName FROM Sales.vSalesPerson")
go func() { rows.Next() }() // DATA RACE
go func() { rows.Next() }() // DATA RACE
// CORRECT: Process rows in the goroutine that created them.
rows, _ := db.QueryContext(ctx, "SELECT BusinessEntityID, FirstName + ' ' + LastName FROM Sales.vSalesPerson")
defer rows.Close()
for rows.Next() {
// Process in this goroutine only.
}
ワーカープールパターン
多くの項目を同時に処理する必要がある場合(例えば数千件のレコードを更新する場合)、固定サイズのワーカープールを使いましょう。 この方法は、プールの枯渇やサーバー過負荷を防ぐために並行性を制限します:
import "sync"
func updateEmployeeLocations(ctx context.Context, db *sql.DB, updates []EmployeeUpdate) error {
const maxWorkers = 10
sem := make(chan struct{}, maxWorkers)
var mu sync.Mutex
var firstErr error
var wg sync.WaitGroup
for _, u := range updates {
select {
case <-ctx.Done():
return ctx.Err()
case sem <- struct{}{}: // Acquire a worker slot.
}
wg.Add(1)
go func(u EmployeeUpdate) {
defer wg.Done()
defer func() { <-sem }() // Release the worker slot.
_, err := db.ExecContext(ctx,
"UPDATE HumanResources.Department SET GroupName = @grp WHERE DepartmentID = @id",
sql.Named("grp", u.GroupName),
sql.Named("id", u.Id))
if err != nil {
mu.Lock()
if firstErr == nil {
firstErr = err
}
mu.Unlock()
}
}(u)
}
wg.Wait()
return firstErr
}
ワーカープールにはerrgroupを使う
golang.org/x/sync/errgroupパッケージは、組み込みのエラー伝播とコンテキストキャンセルによりワーカープールを簡素化します。
import "golang.org/x/sync/errgroup"
func updateDepartmentGroups(ctx context.Context, db *sql.DB, updates []DepartmentUpdate) error {
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(10) // Maximum concurrent goroutines.
for _, u := range updates {
u := u
g.Go(func() error {
_, err := db.ExecContext(ctx,
"UPDATE HumanResources.Department SET GroupName = @grp WHERE DepartmentID = @id",
sql.Named("grp", u.GroupName),
sql.Named("id", u.Id))
return err
})
}
return g.Wait()
}
Tip
エラーグループの限界を MaxOpenConns未満に設定してください。 ワーカー数がプールサイズと等しい場合、ワーカーはすべての接続を消費し、健康チェックやその他のクエリの余地がなくなります。
並列クエリ
独立クエリを同時に実行して総レイテンシを削減します:
func getDashboardData(ctx context.Context, db *sql.DB) (*Dashboard, error) {
g, ctx := errgroup.WithContext(ctx)
var orderCount int
var customerCount int
var revenue float64
g.Go(func() error {
return db.QueryRowContext(ctx,
"SELECT COUNT(*) FROM Sales.SalesOrderHeader WHERE OrderDate >= DATEADD(day, -7, GETUTCDATE())").
Scan(&orderCount)
})
g.Go(func() error {
return db.QueryRowContext(ctx,
"SELECT COUNT(DISTINCT CustomerID) FROM Sales.SalesOrderHeader WHERE OrderDate >= DATEADD(day, -7, GETUTCDATE())").
Scan(&customerCount)
})
g.Go(func() error {
return db.QueryRowContext(ctx,
"SELECT ISNULL(SUM(TotalDue), 0) FROM Sales.SalesOrderHeader WHERE OrderDate >= DATEADD(day, -7, GETUTCDATE())").
Scan(&revenue)
})
if err := g.Wait(); err != nil {
return nil, err
}
return &Dashboard{
OrderCount: orderCount,
CustomerCount: customerCount,
WeeklyRevenue: revenue,
}, nil
}
制御された並行処理を用いたバッチ処理
大規模なバッチ操作(データのインポート、レコード同期)では、バッチ処理と並行性を組み合わせてスループットを最大化します:
func importRecords(ctx context.Context, db *sql.DB, records []Record) error {
const batchSize = 100
const maxWorkers = 5
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(maxWorkers)
for i := 0; i < len(records); i += batchSize {
end := i + batchSize
if end > len(records) {
end = len(records)
}
batch := records[i:end]
g.Go(func() error {
return insertBatch(ctx, db, batch)
})
}
return g.Wait()
}
func insertBatch(ctx context.Context, db *sql.DB, batch []Record) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
stmt, err := tx.Prepare(mssql.CopyIn("Production.ScrapReason", mssql.BulkOptions{}, "Name"))
if err != nil {
return err
}
for _, r := range batch {
if _, err := stmt.Exec(r.Name); err != nil {
return err
}
}
if _, err := stmt.Exec(); err != nil {
return err
}
if err := stmt.Close(); err != nil {
return err
}
return tx.Commit()
}
正常シャットダウン
アプリケーションがシャットダウン信号を受け取ったら、プールを閉じる前にアクティブなデータベース操作を除去してください。
db.Close() を突然呼び出すと、実行中のクエリがキャンセルされ、サーバー側に孤立したセッションが残ることがあります。
import (
"context"
"database/sql"
"log"
"net/http"
"os"
"os/signal"
"syscall"
"time"
)
func main() {
db, err := sql.Open("sqlserver", connString)
if err != nil {
log.Fatal(err)
}
srv := &http.Server{Addr: ":8080"}
// Run the server in a goroutine.
go func() {
if err := srv.ListenAndServe(); err != http.ErrServerClosed {
log.Fatalf("HTTP server error: %v", err)
}
}()
// Wait for interrupt signal.
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Println("Shutting down...")
// Give in-flight HTTP requests up to 30 seconds to complete.
shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
if err := srv.Shutdown(shutdownCtx); err != nil {
log.Printf("HTTP shutdown error: %v", err)
}
// Close the database pool after HTTP handlers have drained.
// This waits for any remaining connections to be returned.
if err := db.Close(); err != nil {
log.Printf("Database close error: %v", err)
}
log.Println("Shutdown complete.")
}
Important
HTTPサーバー(または他のリクエストルーター)のドレインが終わ ったら データベースプールを閉じてください。 プールを先に閉じると、機内担当者が接続エラーを発生させます。
同時ワークロード向けの接続プールサイズ
接続プールのサイズは、アプリケーションの予想される並行性に基づいて決め、ゴルーチンの総数ではなく:
| アプリケーションの種類 | 推奨 MaxOpenConns |
根拠 |
|---|---|---|
| HTTP API、低並行性 | 10-25 | 通常の同時リクエスト数と一致します。 |
| HTTP API、高同時実行性 | 25-50 | パラレルハンドラー用の接続が増えます。 |
| バックグラウンドワーカー(バッチ) | ワーカープールごとに5～10個 | 各労働者には独自のつながりが必要です。 |
| 混合(API + バックグラウンドジョブ) | API + ワーカーニーズの合計 | 各サブシステムに十分な余裕を確保してください。 |
db.SetMaxOpenConns(25) // Total connections across all goroutines.
db.SetMaxIdleConns(10) // Keep warm connections ready for bursts.
db.SetConnMaxLifetime(5 * time.Minute) // Rotate connections for load balancer compatibility.
Tip
常に MaxOpenConns設定します。 デフォルト(0)は無制限です。 負荷がかかる無制限プールは、特に接続制限が階層依存のAzure SQLでは、数百の接続が開かれてサーバーを過負荷にすることがあります。
よくある並行処理のミスを避ける
*sql.Rows を goroutine 間で共有しないでください
*sql.Rows値を複数のゴルーチンに渡すとデータ競合が生じます:
// WRONG: rows is consumed by two goroutines.
rows, _ := db.QueryContext(ctx, "SELECT BusinessEntityID, FirstName + ' ' + LastName FROM Sales.vSalesPerson")
go processRows(rows)
go processRows(rows) // Race condition.
列をループで閉じるのを忘れないでください
ループ内で *sql.Rows をリークすると、接続プールが枯渇します:
// WRONG: Rows leak when the loop starts a new iteration.
for _, location := range locations {
rows, _ := db.QueryContext(ctx,
"SELECT FirstName + ' ' + LastName FROM Sales.vSalesPerson WHERE CountryRegionName = @p1",
sql.Named("p1", location))
for rows.Next() {
// Process...
}
// rows.Close() never called if an error occurs.
}
// CORRECT: Use a helper function with defer.
for _, location := range locations {
if err := processLocation(ctx, db, location); err != nil {
return err
}
}
func processLocation(ctx context.Context, db *sql.DB, location string) error {
rows, err := db.QueryContext(ctx,
"SELECT FirstName + ' ' + LastName FROM Sales.vSalesPerson WHERE CountryRegionName = @p1",
sql.Named("p1", location))
if err != nil {
return err
}
defer rows.Close() // Guaranteed cleanup.
for rows.Next() {
// Process...
}
return rows.Err()
}
接続の固定性が必要な場合を除き、db.Conn は使用しないでください。
db.Conn(ctx) は特定の接続をピン留めします。 不必要に使うと、実効プールサイズが減少します。
// WRONG: Unnecessary pinning.
conn, _ := db.Conn(ctx)
defer conn.Close()
conn.QueryContext(ctx, "SELECT 1") // Use db.QueryContext instead.
// CORRECT: Use db.Conn only for temp tables or session-scoped state.
conn, _ := db.Conn(ctx)
defer conn.Close()
conn.ExecContext(ctx, "CREATE TABLE #Temp (Id INT)")
conn.ExecContext(ctx, "INSERT INTO #Temp VALUES (1)")
conn.QueryContext(ctx, "SELECT * FROM #Temp")
並行性チェックリスト
| Area | レコメンデーション |
|---|---|
| プールシェアリング | アプリケーション全体に対して1つの *sql.DB インスタンスを作成します。 |
| Goroutineの安全性 |
*sql.Rows、*sql.Tx、*sql.Connをゴルーチン間で共有しないでください。 |
| ワーカープール |
errgroup.SetLimitまたはセマフォチャネルを使って並行性を制御してください。 |
| プールのサイズ分け | サーバー接続 MaxOpenConns 制限より低く設定してください。 |
| 正常シャットダウン | データベースプールを閉じる前にHTTPハンドラーを排出してください。 |
| リソースのクリーンアップ | 必ず、それらを作成した goroutine で defer rows.Close() と defer tx.Rollback() を行ってください。 |
| 並列クエリ | 集約ページの遅延を減らすために、独立したクエリを同時に実行します。 |