Go y PostgreSQL: técnicas avanzadas con pgx, pgxpool y transacciones

Introducción: por qué pgx en lugar de database/sql

La interfaz estándar database/sql de Go es universal, pero precisamente esa universalidad se convierte en un cuello de botella cuando se trabaja seriamente con PostgreSQL. La biblioteca pgx de jackc proporciona acceso directo al protocolo de PostgreSQL, sin pasar por las abstracciones de database/sql, lo que ofrece varias ventajas fundamentales.

Soporte nativo de tipos PostgreSQL: JSONB, arrays, UUID, pgtype.Numeric sin conversiones innecesarias.
Soporte de consultas batch mediante SendBatch, lo que reduce drásticamente el número de round-trips.
Pool de conexiones integrado pgxpool con configuración detallada de health-check y timeouts.
Soporte del protocolo COPY para inserción masiva de datos.
Trabajo directo con prepared statements a nivel de protocolo.

Para desarrolladores Go que construyen servicios de alta carga sobre PostgreSQL, migrar de database/sql + lib/pq a pgx v5 es una de las decisiones técnicas más rentables de 2024–2026. Instalación:

go get github.com/jackc/pgx/v5\ngo get github.com/jackc/pgx/v5/pgxpool

Configuración de pgxpool: tamaño del pool, timeouts, health check

El pool de conexiones es el elemento central para un rendimiento óptimo con PostgreSQL desde Go. pgxpool gestiona el ciclo de vida de las conexiones, su reutilización y la verificación de disponibilidad.

package main\n\nimport (\n    \"context\"\n    \"fmt\"\n    \"log\"\n    \"time\"\n\n    \"github.com/jackc/pgx/v5/pgxpool\"\n)\n\nfunc NewPool(dsn string) (*pgxpool.Pool, error) {\n    config, err := pgxpool.ParseConfig(dsn)\n    if err != nil {\n        return nil, fmt.Errorf(\"parse config: %w\", err)\n    }\n\n    // Número máximo de conexiones en el pool\n    config.MaxConns = 30\n    // Número mínimo de conexiones inactivas\n    config.MinConns = 5\n    // Tiempo de vida máximo de una conexión\n    config.MaxConnLifetime = 1 * time.Hour\n    // Tiempo máximo de inactividad de una conexión\n    config.MaxConnIdleTime = 30 * time.Minute\n    // Período de health check\n    config.HealthCheckPeriod = 1 * time.Minute\n    // Timeout para establecer una conexión\n    config.ConnConfig.ConnectTimeout = 5 * time.Second\n\n    // Health check: ejecutamos SELECT 1 al obtener una conexión\n    config.BeforeAcquire = func(ctx context.Context, conn *pgxpool.Conn) bool {\n        return conn.Ping(ctx) == nil\n    }\n\n    // Hook tras liberar una conexión\n    config.AfterRelease = func(conn *pgx.Conn) bool {\n        // Reseteamos prepared statements si la conexión tuvo errores\n        return conn.IsClosed() == false\n    }\n\n    pool, err := pgxpool.NewWithConfig(context.Background(), config)\n    if err != nil {\n        return nil, fmt.Errorf(\"create pool: %w\", err)\n    }\n\n    return pool, nil\n}

Algunas reglas importantes al configurar pgxpool:

MaxConns no debe superar max_connections de PostgreSQL menos las conexiones para replicación y tareas administrativas. Regla práctica: MaxConns = (número de núcleos CPU * 2) + número de discos.
MinConns permite mantener conexiones «calientes» y evitar latencia ante picos de tráfico.
MaxConnLifetime previene la acumulación de conexiones de larga duración que pueden retener recursos en el lado de PostgreSQL.
No abuses de BeforeAcquire con Ping: añade latencia en cada obtención de conexión. Úsalo solo si hay problemas con conexiones colgadas.

Trabajo con transacciones: transacciones explícitas, savepoints, manejo de errores

Las transacciones en PostgreSQL a través de pgx requieren gestión explícita. Un error frecuente es ignorar el rollback ante un panic o un error.

func TransferFunds(ctx context.Context, pool *pgxpool.Pool, fromID, toID int64, amount float64) error {\n    tx, err := pool.Begin(ctx)\n    if err != nil {\n        return fmt.Errorf(\"begin tx: %w\", err)\n    }\n    // Garantizamos el rollback al salir de la función con error\n    defer func() {\n        if err != nil {\n            _ = tx.Rollback(ctx)\n        }\n    }()\n\n    // Débito\n    _, err = tx.Exec(ctx,\n        \"UPDATE accounts SET balance = balance - $1 WHERE id = $2 AND balance >= $1\",\n        amount, fromID,\n    )\n    if err != nil {\n        return fmt.Errorf(\"debit: %w\", err)\n    }\n\n    // Savepoint antes del crédito\n    _, err = tx.Exec(ctx, \"SAVEPOINT before_credit\")\n    if err != nil {\n        return fmt.Errorf(\"savepoint: %w\", err)\n    }\n\n    // Crédito\n    _, err = tx.Exec(ctx,\n        \"UPDATE accounts SET balance = balance + $1 WHERE id = $2\",\n        amount, toID,\n    )\n    if err != nil {\n        // Rollback al savepoint, no al inicio de la transacción\n        if rbErr := tx.Exec(ctx, \"ROLLBACK TO SAVEPOINT before_credit\"); rbErr != nil {\n            return fmt.Errorf(\"rollback to savepoint: %w\", rbErr)\n        }\n        return fmt.Errorf(\"credit: %w\", err)\n    }\n\n    // Registramos la operación en la misma transacción\n    _, err = tx.Exec(ctx,\n        \"INSERT INTO audit_log (from_id, to_id, amount, created_at) VALUES ($1, $2, $3, NOW())\",\n        fromID, toID, amount,\n    )\n    if err != nil {\n        return fmt.Errorf(\"audit log: %w\", err)\n    }\n\n    return tx.Commit(ctx)\n}

Para los niveles de aislamiento, usa pool.BeginTx con indicación explícita del nivel:

tx, err := pool.BeginTx(ctx, pgx.TxOptions{\n    IsoLevel:   pgx.Serializable,\n    AccessMode: pgx.ReadWrite,\n})

El patrón «función con transacción» es un wrapper conveniente para reutilizar:

func WithTx(ctx context.Context, pool *pgxpool.Pool, fn func(pgx.Tx) error) error {\n    tx, err := pool.Begin(ctx)\n    if err != nil {\n        return err\n    }\n    defer func() {\n        if p := recover(); p != nil {\n            _ = tx.Rollback(ctx)\n            panic(p)\n        } else if err != nil {\n            _ = tx.Rollback(ctx)\n        } else {\n            err = tx.Commit(ctx)\n        }\n    }()\n    err = fn(tx)\n    return err\n}

Prepared statements y su impacto en el rendimiento

pgx almacena automáticamente en caché los prepared statements en el modo de protocolo de consulta extendida. En la primera ejecución, la consulta se analiza y planifica en el lado de PostgreSQL; las llamadas posteriores utilizan el plan en caché.

// Preparación explícita del statement\nfunc PrepareStatements(ctx context.Context, conn *pgx.Conn) error {\n    _, err := conn.Prepare(ctx, \"get_user_by_email\",\n        \"SELECT id, name, email, created_at FROM users WHERE email = $1 AND deleted_at IS NULL\",\n    )\n    if err != nil {\n        return fmt.Errorf(\"prepare get_user_by_email: %w\", err)\n    }\n    return nil\n}\n\n// Uso del statement preparado\nfunc GetUserByEmail(ctx context.Context, conn *pgx.Conn, email string) (*User, error) {\n    var u User\n    err := conn.QueryRow(ctx, \"get_user_by_email\", email).Scan(\n        &u.ID, &u.Name, &u.Email, &u.CreatedAt,\n    )\n    if err != nil {\n        return nil, err\n    }\n    return &u, nil\n}

Un matiz importante con pgxpool: los prepared statements están vinculados a una conexión concreta. Al usar el pool, se recomienda utilizar el caché automático mediante QueryExecModeSimpleProtocol o la configuración de StatementCacheCapacity en la configuración de la conexión:

config.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeCacheStatement\nconfig.ConnConfig.StatementCacheCapacity = 512

Trabajo con tipos especiales de PostgreSQL: JSONB, arrays, UUID

Una de las principales ventajas de pgx es el trabajo nativo con tipos de PostgreSQL sin serializaciones innecesarias.

JSONB

import \"github.com/jackc/pgx/v5/pgtype\"\n\ntype UserSettings struct {\n    Theme    string `json:\"theme\"`\n    Language string `json:\"language\"`\n    Notifications bool `json:\"notifications\"`\n}\n\nfunc SaveUserSettings(ctx context.Context, pool *pgxpool.Pool, userID int64, settings UserSettings) error {\n    data, err := json.Marshal(settings)\n    if err != nil {\n        return err\n    }\n    _, err = pool.Exec(ctx,\n        \"UPDATE users SET settings = $1 WHERE id = $2\",\n        data, userID,\n    )\n    return err\n}\n\nfunc GetUserSettings(ctx context.Context, pool *pgxpool.Pool, userID int64) (*UserSettings, error) {\n    var raw []byte\n    err := pool.QueryRow(ctx,\n        \"SELECT settings FROM users WHERE id = $1\",\n        userID,\n    ).Scan(&raw)\n    if err != nil {\n        return nil, err\n    }\n    var s UserSettings\n    if err := json.Unmarshal(raw, &s); err != nil {\n        return nil, err\n    }\n    return &s, nil\n}

Arrays de PostgreSQL

// Inserción de un array de etiquetas\nfunc AddTags(ctx context.Context, pool *pgxpool.Pool, articleID int64, tags []string) error {\n    _, err := pool.Exec(ctx,\n        \"UPDATE articles SET tags = $1::text[] WHERE id = $2\",\n        tags, articleID,\n    )\n    return err\n}\n\n// Lectura de un array\nfunc GetTags(ctx context.Context, pool *pgxpool.Pool, articleID int64) ([]string, error) {\n    var tags []string\n    err := pool.QueryRow(ctx,\n        \"SELECT tags FROM articles WHERE id = $1\",\n        articleID,\n    ).Scan(&tags)\n    return tags, err\n}

UUID

import \"github.com/google/uuid\"\n\nfunc CreateOrder(ctx context.Context, pool *pgxpool.Pool, userID int64) (uuid.UUID, error) {\n    id := uuid.New()\n    _, err := pool.Exec(ctx,\n        \"INSERT INTO orders (id, user_id, status, created_at) VALUES ($1, $2, 'pending', NOW())\",\n        id, userID,\n    )\n    if err != nil {\n        return uuid.Nil, err\n    }\n    return id, nil\n}

Consultas batch con pgx: reducción de round-trips

El batch es la herramienta más poderosa de pgx para escenarios de alta carga. En lugar de N consultas separadas, enviamos un único batch y leemos N resultados en un solo round-trip.

func GetMultipleUsers(ctx context.Context, pool *pgxpool.Pool, ids []int64) ([]*User, error) {\n    conn, err := pool.Acquire(ctx)\n    if err != nil {\n        return nil, err\n    }\n    defer conn.Release()\n\n    batch := &pgx.Batch{}\n    for _, id := range ids {\n        batch.Queue(\"SELECT id, name, email FROM users WHERE id = $1\", id)\n    }\n\n    results := conn.SendBatch(ctx, batch)\n    defer results.Close()\n\n    var users []*User\n    for range ids {\n        var u User\n        err := results.QueryRow().Scan(&u.ID, &u.Name, &u.Email)\n        if err != nil {\n            return nil, fmt.Errorf(\"scan user: %w\", err)\n        }\n        users = append(users, &u)\n    }\n\n    return users, nil\n}\n\n// Inserción batch\nfunc BulkInsertEvents(ctx context.Context, pool *pgxpool.Pool, events []Event) error {\n    conn, err := pool.Acquire(ctx)\n    if err != nil {\n        return err\n    }\n    defer conn.Release()\n\n    batch := &pgx.Batch{}\n    for _, e := range events {\n        batch.Queue(\n            \"INSERT INTO events (user_id, type, payload, created_at) VALUES ($1, $2, $3, $4)\",\n            e.UserID, e.Type, e.Payload, e.CreatedAt,\n        )\n    }\n\n    br := conn.SendBatch(ctx, batch)\n    defer br.Close()\n\n    for i := range events {\n        if _, err := br.Exec(); err != nil {\n            return fmt.Errorf(\"insert event %d: %w\", i, err)\n        }\n    }\n    return nil\n}

Para la inserción masiva de cientos de miles de filas, usa el protocolo COPY: es incluso más rápido que las consultas batch:

func BulkCopyUsers(ctx context.Context, pool *pgxpool.Pool, users []User) error {\n    conn, err := pool.Acquire(ctx)\n    if err != nil {\n        return err\n    }\n    defer conn.Release()\n\n    rows := make([][]interface{}, len(users))\n    for i, u := range users {\n        rows[i] = []interface{}{u.Name, u.Email, u.CreatedAt}\n    }\n\n    _, err = conn.Conn().CopyFrom(\n        ctx,\n        pgx.Identifier{\"users\"},\n        []string{\"name\", \"email\", \"created_at\"},\n        pgx.CopyFromRows(rows),\n    )\n    return err\n}

Actualizaciones concurrentes: bloqueos optimistas y SELECT FOR UPDATE

En sistemas con alta concurrencia, es fundamental manejar correctamente las actualizaciones paralelas. pgx ofrece herramientas convenientes para ambos enfoques.

SELECT FOR UPDATE (bloqueo pesimista)

func ReserveProduct(ctx context.Context, pool *pgxpool.Pool, productID int64, quantity int) error {\n    return WithTx(ctx, pool, func(tx pgx.Tx) error {\n        var stock int\n        err := tx.QueryRow(ctx,\n            \"SELECT stock FROM products WHERE id = $1 FOR UPDATE\",\n            productID,\n        ).Scan(&stock)\n        if err != nil {\n            return fmt.Errorf(\"lock product: %w\", err)\n        }\n\n        if stock < quantity {\n            return fmt.Errorf(\"stock insuficiente: disponible %d, solicitado %d\", stock, quantity)\n        }\n\n        _, err = tx.Exec(ctx,\n            \"UPDATE products SET stock = stock - $1 WHERE id = $2\",\n            quantity, productID,\n        )\n        return err\n    })\n}

Bloqueos optimistas mediante version/updated_at

type Product struct {\n    ID      int64\n    Name    string\n    Price   float64\n    Version int // Contador de versiones\n}\n\nfunc UpdateProductOptimistic(ctx context.Context, pool *pgxpool.Pool, p Product) error {\n    result, err := pool.Exec(ctx,\n        `UPDATE products \n         SET name = $1, price = $2, version = version + 1 \n         WHERE id = $3 AND version = $4`,\n        p.Name, p.Price, p.ID, p.Version,\n    )\n    if err != nil {\n        return fmt.Errorf(\"update: %w\", err)\n    }\n\n    if result.RowsAffected() == 0 {\n        return fmt.Errorf(\"conflicto de bloqueo optimista: el producto %d fue modificado concurrentemente\", p.ID)\n    }\n    return nil\n}\n\n// Wrapper con reintentos para bloqueos optimistas\nfunc WithOptimisticRetry(maxAttempts int, fn func() error) error {\n    for i := 0; i < maxAttempts; i++ {\n        err := fn()\n        if err == nil {\n            return nil\n        }\n        // Reintentamos solo ante conflicto de versiones\n        if strings.Contains(err.Error(), \"optimistic lock conflict\") {\n            time.Sleep(time.Duration(i*10) * time.Millisecond)\n            continue\n        }\n        return err\n    }\n    return fmt.Errorf(\"se superó el máximo de reintentos (%d)\", maxAttempts)\n}

Monitoreo del pool de conexiones y diagnóstico de cuellos de botella

pgxpool proporciona el método Stat() para obtener el estado actual del pool. Intégralo con Prometheus o cualquier otro sistema de monitoreo.

import (\n    \"github.com/prometheus/client_golang/prometheus\"\n    \"github.com/prometheus/client_golang/prometheus/promauto\"\n)\n\nvar (\n    poolAcquiredConns = promauto.NewGauge(prometheus.GaugeOpts{\n        Name: \"pgxpool_acquired_connections\",\n        Help: \"Number of currently acquired connections\",\n    })\n    poolIdleConns = promauto.NewGauge(prometheus.GaugeOpts{\n        Name: \"pgxpool_idle_connections\",\n        Help: \"Number of idle connections in the pool\",\n    })\n    poolTotalConns = promauto.NewGauge(prometheus.GaugeOpts{\n        Name: \"pgxpool_total_connections\",\n        Help: \"Total number of connections in the pool\",\n    })\n    poolWaitCount = promauto.NewCounter(prometheus.CounterOpts{\n        Name: \"pgxpool_wait_total\",\n        Help: \"Total number of times waited for a connection\",\n    })\n)\n\nfunc MonitorPool(pool *pgxpool.Pool, interval time.Duration) {\n    ticker := time.NewTicker(interval)\n    defer ticker.Stop()\n    for range ticker.C {\n        stat := pool.Stat()\n        poolAcquiredConns.Set(float64(stat.AcquiredConns()))\n        poolIdleConns.Set(float64(stat.IdleConns()))\n        poolTotalConns.Set(float64(stat.TotalConns()))\n        poolWaitCount.Add(float64(stat.EmptyAcquireCount()))\n    }\n}

Métricas clave para analizar el rendimiento de PostgreSQL desde una aplicación Go:

AcquiredConns / MaxConns: si se acerca al 100%, aumenta el pool u optimiza las consultas.
EmptyAcquireCount: número de esperas por una conexión libre. El crecimiento de esta métrica indica un pool insuficiente.
MaxConnLifetimeDestroyCount: la recreación frecuente de conexiones por lifetime puede indicar un MaxConnLifetime demasiado corto.

Para diagnosticar consultas lentas, usa pg_stat_statements en el lado de PostgreSQL y el logging de slow queries en el lado de Go:

// Middleware para registrar consultas lentas\ntype LoggingQuerier struct {\n    pool      *pgxpool.Pool\n    threshold time.Duration\n    logger    *slog.Logger\n}\n\nfunc (lq *LoggingQuerier) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {\n    start := time.Now()\n    row := lq.pool.QueryRow(ctx, sql, args...)\n    elapsed := time.Since(start)\n    if elapsed > lq.threshold {\n        lq.logger.WarnContext(ctx, \"slow query\",\n            \"sql\", sql,\n            \"duration_ms\", elapsed.Milliseconds(),\n        )\n    }\n    return row\n}

Integración con Docker para desarrollo local

Para un desarrollo local reproducible con PostgreSQL, usa Docker Compose. A continuación, una configuración mínima con ajustes de rendimiento.

version: '3.8'\nservices:\n  postgres:\n    image: postgres:16-alpine\n    environment:\n      POSTGRES_DB: myapp\n      POSTGRES_USER: myapp\n      POSTGRES_PASSWORD: secret\n    ports:\n      - \"5432:5432\"\n    volumes:\n      - postgres_data:/var/lib/postgresql/data\n      - ./init.sql:/docker-entrypoint-initdb.d/init.sql\n    command: >\n      postgres\n        -c max_connections=200\n        -c shared_buffers=256MB\n        -c effective_cache_size=768MB\n        -c work_mem=4MB\n        -c log_min_duration_statement=100\n        -c log_statement=all\n    healthcheck:\n      test: [\"CMD-SHELL\", \"pg_isready -U myapp\"]\n      interval: 5s\n      timeout: 5s\n      retries: 5\n\nvolumes:\n  postgres_data:

En el código Go, usa variables de entorno para el DSN, de modo que la configuración funcione tanto en local como en producción:

func main() {\n    dsn := os.Getenv(\"DATABASE_URL\")\n    if dsn == \"\" {\n        dsn = \"postgres://myapp:secret@localhost:5432/myapp?sslmode=disable\"\n    }\n\n    pool, err := NewPool(dsn)\n    if err != nil {\n        log.Fatalf(\"failed to create pool: %v\", err)\n    }\n    defer pool.Close()\n\n    // Verificamos la conexión al arrancar\n    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)\n    defer cancel()\n\n    if err := pool.Ping(ctx); err != nil {\n        log.Fatalf(\"failed to ping database: %v\", err)\n    }\n    log.Println(\"Connected to PostgreSQL\")\n}

Conclusión y buenas prácticas

Trabajar con PostgreSQL desde Go usando pgx no es solo elegir un driver, sino toda una ecosistema de herramientas para construir servicios fiables y de alto rendimiento. Resumamos las prácticas clave:

Usa siempre pgxpool en código de producción. Las conexiones directas con pgx.Connect son adecuadas solo para tareas administrativas y pruebas.
Configura MaxConns de forma consciente: guíate por las capacidades del servidor PostgreSQL, no por números arbitrarios.
Defer rollback siempre: el patrón defer tx.Rollback() es seguro, se ignora tras un Commit exitoso.
Usa consultas batch para operaciones que puedan agruparse. El ahorro en round-trips es especialmente notable con alta latencia de red.
Tipos nativos en lugar de strings: UUID, JSONB, arrays — úsalos directamente, no los conviertas a string sin necesidad.
Monitorea el pool: las métricas de pgxpool deben ser parte de la observabilidad de tu servicio desde el primer día.
Protocolo COPY para bulk insert: si necesitas insertar miles de filas, COPY es entre 5 y 10 veces más rápido que un INSERT batch.
Savepoints para rollbacks parciales en transacciones complejas: es una herramienta estándar de PostgreSQL, no la subestimes.
Registra las consultas lentas: configura log_min_duration_statement en PostgreSQL y añade middleware en Go para correlacionarlas.
Docker para desarrollo local con healthcheck garantiza que la aplicación arranque solo cuando la base de datos esté lista.

pgx se desarrolla activamente: v5 trajo una API mejorada para el trabajo con tipos, gestión de consultas más flexible y mejor rendimiento. Mantente al día con las versiones y actualiza las dependencias: es una inversión directa en el rendimiento de tu servicio Go sobre PostgreSQL.

Go y PostgreSQL: técnicas avanzadas con pgx, pool de conexiones y transacciones