Replicação
Cópias do banco para leitura e para tolerar falhas, e o atraso entre a principal e as réplicas.
O código desta página foi executado nas cinco linguagens contra um PostgreSQL 18 com um principal e uma réplica em replicação por fluxo. As medições do nível Conceito vêm do mesmo par.
Ler o que acabou de gravar, em cada linguagem
Cada versão tem duas funções. criarPedido grava no principal e, num segundo comando, lê a posição atual do log (pg_current_wal_lsn()). buscarPedido recebe esse LSN e pergunta à réplica se ela já aplicou o log até ali (pg_last_wal_replay_lsn()): se sim, lê da réplica; se não, lê do principal.
O teste chama as funções três vezes: com a réplica em dia; com a aplicação do log na réplica pausada (pg_wal_replay_pause()), passando o LSN; e com a réplica pausada, sem o LSN.
import pg from "pg";
const principal = new pg.Pool({ connectionString: process.env.URL_PRINCIPAL });
const replica = new pg.Pool({ connectionString: process.env.URL_REPLICA });
// Grava no principal e devolve a posição do log (LSN) depois da confirmação
async function criarPedido(total: number) {
const { rows } = await principal.query(
"INSERT INTO pedidos (total) VALUES ($1) RETURNING id", [total]);
// Num comando separado: dentro do INSERT, o LSN ainda não inclui o COMMIT
const lsn = await principal.query("SELECT pg_current_wal_lsn()::text AS lsn");
return { id: Number(rows[0].id), lsn: lsn.rows[0].lsn as string };
}
// Lê da réplica só se ela já aplicou a gravação; senão, do principal
async function buscarPedido(id: number, lsnMinimo?: string) {
if (lsnMinimo) {
const { rows } = await replica.query(
"SELECT pg_last_wal_replay_lsn() >= $1::pg_lsn AS em_dia", [lsnMinimo]);
if (!rows[0].em_dia) {
const r = await principal.query("SELECT id, total FROM pedidos WHERE id = $1", [id]);
return { origem: "principal", pedido: r.rows[0] };
}
}
const r = await replica.query("SELECT id, total FROM pedidos WHERE id = $1", [id]);
return { origem: "réplica", pedido: r.rows[0] };
}
// Teste: réplica em dia, e depois com a aplicação do log pausada
const a = await criarPedido(10);
await new Promise((r) => setTimeout(r, 300));
console.log("réplica em dia:", JSON.stringify(await buscarPedido(a.id, a.lsn)));
await replica.query("SELECT pg_wal_replay_pause()");
const b = await criarPedido(20);
console.log("réplica pausada, com LSN:", JSON.stringify(await buscarPedido(b.id, b.lsn)));
console.log("réplica pausada, sem LSN:", JSON.stringify(await buscarPedido(b.id)));
await replica.query("SELECT pg_wal_replay_resume()");
await principal.end();
await replica.end();import json
import os
import time
import psycopg
principal = psycopg.connect(os.environ["URL_PRINCIPAL"], autocommit=True)
replica = psycopg.connect(os.environ["URL_REPLICA"], autocommit=True)
def criar_pedido(total: float) -> tuple[int, str]:
"""Grava no principal e devolve a posição do log (LSN) depois da confirmação."""
(id_,) = principal.execute(
"INSERT INTO pedidos (total) VALUES (%s) RETURNING id", (total,)).fetchone()
# Num comando separado: dentro do INSERT, o LSN ainda não inclui o COMMIT
(lsn,) = principal.execute("SELECT pg_current_wal_lsn()::text").fetchone()
return id_, lsn
def buscar_pedido(id_: int, lsn_minimo: str | None = None) -> dict:
"""Lê da réplica só se ela já aplicou a gravação; senão, do principal."""
banco, origem = replica, "réplica"
if lsn_minimo:
(em_dia,) = replica.execute(
"SELECT pg_last_wal_replay_lsn() >= %s::pg_lsn", (lsn_minimo,)).fetchone()
if not em_dia:
banco, origem = principal, "principal"
linha = banco.execute("SELECT id, total FROM pedidos WHERE id = %s", (id_,)).fetchone()
return {"origem": origem, "pedido": linha and {"id": linha[0], "total": str(linha[1])}}
# Teste: réplica em dia, e depois com a aplicação do log pausada
a = criar_pedido(10)
time.sleep(0.3)
print("réplica em dia:", json.dumps(buscar_pedido(*a), ensure_ascii=False))
replica.execute("SELECT pg_wal_replay_pause()")
b = criar_pedido(20)
print("réplica pausada, com LSN:", json.dumps(buscar_pedido(*b), ensure_ascii=False))
print("réplica pausada, sem LSN:", json.dumps(buscar_pedido(b[0]), ensure_ascii=False))
replica.execute("SELECT pg_wal_replay_resume()")package main
import (
"context"
"fmt"
"os"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
type Bancos struct{ Principal, Replica *pgxpool.Pool }
// criarPedido grava no principal e devolve a posição do log (LSN) depois da confirmação
func (b Bancos) criarPedido(ctx context.Context, total float64) (id int64, lsn string, err error) {
err = b.Principal.QueryRow(ctx,
"INSERT INTO pedidos (total) VALUES ($1) RETURNING id", total).Scan(&id)
if err != nil {
return
}
// Num comando separado: dentro do INSERT, o LSN ainda não inclui o COMMIT
err = b.Principal.QueryRow(ctx, "SELECT pg_current_wal_lsn()::text").Scan(&lsn)
return
}
// buscarPedido lê da réplica só se ela já aplicou a gravação; senão, do principal
func (b Bancos) buscarPedido(
ctx context.Context, id int64, lsnMinimo string,
) (string, float64, error) {
banco, origem := b.Replica, "réplica"
if lsnMinimo != "" {
var emDia bool
err := b.Replica.QueryRow(ctx,
"SELECT pg_last_wal_replay_lsn() >= $1::pg_lsn", lsnMinimo).Scan(&emDia)
if err != nil {
return "", 0, err
}
if !emDia {
banco, origem = b.Principal, "principal"
}
}
var total float64
err := banco.QueryRow(ctx, "SELECT total::float8 FROM pedidos WHERE id = $1", id).Scan(&total)
if err == pgx.ErrNoRows {
return origem, 0, fmt.Errorf("pedido %d não encontrado na %s", id, origem)
}
return origem, total, err
}
func main() {
ctx := context.Background()
p, _ := pgxpool.New(ctx, os.Getenv("URL_PRINCIPAL"))
r, _ := pgxpool.New(ctx, os.Getenv("URL_REPLICA"))
defer p.Close()
defer r.Close()
b := Bancos{p, r}
// Teste: réplica em dia, e depois com a aplicação do log pausada
id, lsn, _ := b.criarPedido(ctx, 10)
time.Sleep(300 * time.Millisecond)
fmt.Println(b.buscarPedido(ctx, id, lsn))
r.Exec(ctx, "SELECT pg_wal_replay_pause()")
id, lsn, _ = b.criarPedido(ctx, 20)
fmt.Println(b.buscarPedido(ctx, id, lsn))
fmt.Println(b.buscarPedido(ctx, id, ""))
r.Exec(ctx, "SELECT pg_wal_replay_resume()")
}import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
public class Rotas {
record Gravado(long id, String lsn) {}
record Lido(String origem, BigDecimal total) {}
final Connection principal;
final Connection replica;
Rotas(Connection principal, Connection replica) {
this.principal = principal;
this.replica = replica;
}
// Grava no principal e devolve a posição do log (LSN) depois da confirmação
Gravado criarPedido(BigDecimal total) throws SQLException {
long id;
var sql = "INSERT INTO pedidos (total) VALUES (?) RETURNING id";
try (var c = principal.prepareStatement(sql)) {
c.setBigDecimal(1, total);
try (var r = c.executeQuery()) {
r.next();
id = r.getLong(1);
}
}
// Num comando separado: dentro do INSERT, o LSN ainda não inclui o COMMIT
try (var c = principal.createStatement();
var r = c.executeQuery("SELECT pg_current_wal_lsn()::text")) {
r.next();
return new Gravado(id, r.getString(1));
}
}
// Lê da réplica só se ela já aplicou a gravação; senão, do principal
Lido buscarPedido(long id, String lsnMinimo) throws SQLException {
var banco = replica;
var origem = "réplica";
if (lsnMinimo != null) {
try (var c = replica.prepareStatement("SELECT pg_last_wal_replay_lsn() >= ?::pg_lsn")) {
c.setString(1, lsnMinimo);
try (var r = c.executeQuery()) {
r.next();
if (!r.getBoolean(1)) {
banco = principal;
origem = "principal";
}
}
}
}
try (var c = banco.prepareStatement("SELECT total FROM pedidos WHERE id = ?")) {
c.setLong(1, id);
try (var r = c.executeQuery()) {
return new Lido(origem, r.next() ? r.getBigDecimal(1) : null);
}
}
}
public static void main(String[] args) throws Exception {
try (var p = DriverManager.getConnection(System.getenv("JDBC_PRINCIPAL"));
var r = DriverManager.getConnection(System.getenv("JDBC_REPLICA"))) {
var rotas = new Rotas(p, r);
// Teste: réplica em dia, e depois com a aplicação do log pausada
var a = rotas.criarPedido(new BigDecimal("10"));
Thread.sleep(300);
System.out.println("réplica em dia: " + rotas.buscarPedido(a.id(), a.lsn()));
r.createStatement().execute("SELECT pg_wal_replay_pause()");
var b = rotas.criarPedido(new BigDecimal("20"));
System.out.println("réplica pausada, com LSN: " + rotas.buscarPedido(b.id(), b.lsn()));
System.out.println("réplica pausada, sem LSN: " + rotas.buscarPedido(b.id(), null));
r.createStatement().execute("SELECT pg_wal_replay_resume()");
}
}
}using Npgsql;
var urlPrincipal = Environment.GetEnvironmentVariable("URL_PRINCIPAL")!;
var urlReplica = Environment.GetEnvironmentVariable("URL_REPLICA")!;
await using var principal = NpgsqlDataSource.Create(urlPrincipal);
await using var replica = NpgsqlDataSource.Create(urlReplica);
// Grava no principal e devolve a posição do log (LSN) depois da confirmação
async Task<(long Id, string Lsn)> CriarPedido(decimal total)
{
await using var c = principal.CreateCommand(
"INSERT INTO pedidos (total) VALUES ($1) RETURNING id");
c.Parameters.Add(new NpgsqlParameter { Value = total });
var id = (long)(await c.ExecuteScalarAsync())!;
// Num comando separado: dentro do INSERT, o LSN ainda não inclui o COMMIT
await using var l = principal.CreateCommand("SELECT pg_current_wal_lsn()::text");
return (id, (string)(await l.ExecuteScalarAsync())!);
}
// Lê da réplica só se ela já aplicou a gravação; senão, do principal
async Task<(string Origem, decimal? Total)> BuscarPedido(long id, string? lsnMinimo = null)
{
var (banco, origem) = (replica, "réplica");
if (lsnMinimo is not null)
{
await using var c = replica.CreateCommand("SELECT pg_last_wal_replay_lsn() >= $1::pg_lsn");
c.Parameters.Add(new NpgsqlParameter { Value = lsnMinimo });
if (!(bool)(await c.ExecuteScalarAsync())!) (banco, origem) = (principal, "principal");
}
await using var consulta = banco.CreateCommand("SELECT total FROM pedidos WHERE id = $1");
consulta.Parameters.Add(new NpgsqlParameter { Value = id });
return (origem, (decimal?)await consulta.ExecuteScalarAsync());
}
// Teste: réplica em dia, e depois com a aplicação do log pausada
var a = await CriarPedido(10);
await Task.Delay(300);
Console.WriteLine($"réplica em dia: {await BuscarPedido(a.Id, a.Lsn)}");
await using (var pausa = replica.CreateCommand("SELECT pg_wal_replay_pause()"))
await pausa.ExecuteNonQueryAsync();
var b = await CriarPedido(20);
Console.WriteLine($"réplica pausada, com LSN: {await BuscarPedido(b.Id, b.Lsn)}");
Console.WriteLine($"réplica pausada, sem LSN: {await BuscarPedido(b.Id)}");
await using (var volta = replica.CreateCommand("SELECT pg_wal_replay_resume()"))
await volta.ExecuteNonQueryAsync();O resultado, nas cinco linguagens
| Situação | Leu de | Achou o pedido? |
|---|---|---|
| Réplica em dia, com o LSN | réplica | sim |
| Réplica pausada, com o LSN | principal | sim |
| Réplica pausada, sem o LSN | réplica | não |
A terceira linha é o problema que a técnica resolve: sem conferir o LSN, a leitura logo depois da gravação não encontrou o pedido recém-criado. O resultado foi o mesmo nas cinco linguagens.
O que muda de uma linguagem para outra
| Dois bancos | Comparar LSNs | |
|---|---|---|
| TypeScript | Dois pg.Pool | O LSN vai como texto e é convertido com $1::pg_lsn |
| Python | Duas conexões (ou dois pools) | Igual, com %s::pg_lsn |
| Go | Uma struct com dois pgxpool.Pool | Igual, com $1::pg_lsn |
| Java | Duas Connection (ou dois DataSource) | Igual, com ?::pg_lsn |
| C# | Dois NpgsqlDataSource | Igual, com $1::pg_lsn |
A comparação fica no banco, com o tipo pg_lsn: o LSN tem o formato 0/3000148, e comparar como texto daria resultados errados. Num sistema real, o LSN da última gravação do usuário fica na sessão ou num cookie, e vale para todas as leituras dele nos segundos seguintes.
Frameworks e ORMs têm configurações para separar leituras e gravações (réplicas no Django e no Rails, DbContext separados no EF Core, AbstractRoutingDataSource no Spring). Eles roteiam por tipo de operação, e o "ler o que gravou" costuma ficar por conta da aplicação.
Montar a réplica
Os comandos usados no teste, no PostgreSQL 18:
# No principal: um usuário de replicação e um slot para a réplica
psql -c "CREATE ROLE replicador WITH REPLICATION LOGIN PASSWORD '...'"
psql -c "SELECT pg_create_physical_replication_slot('replica1')"
# e, no pg_hba.conf: host replication replicador <rede da réplica> scram-sha-256
# Na máquina da réplica: copia o principal e grava a configuração de réplica (-R)
pg_basebackup -h principal -U replicador -D "$PGDATA" -R -S replica1 -X streamE as configurações que o nível Conceito testou, no principal:
synchronous_standby_names = 'walreceiver' # no teste; com quórum: 'ANY 1 (r1, r2)'
max_slot_wal_keep_size = '256MB' # limite de WAL retido por slotPara conferir o estado:
-- No principal
SELECT application_name, state, sync_state, replay_lag FROM pg_stat_replication;
SELECT slot_name, active, wal_status FROM pg_replication_slots;
-- Na réplica
SELECT pg_is_in_recovery(), pg_last_wal_replay_lsn(), now() - pg_last_xact_replay_timestamp() AS atraso;Lista de revisão
Escolha
- O time sabe quanto se perde numa troca de principal no modo de replicação escolhido?
- A replicação síncrona usa quórum de duas ou mais réplicas, e nunca uma única?
- Há uma réplica em outra zona, e um plano para desastres regionais?
Troca de principal
- A troca é automática, com consenso, ou feita por um banco gerenciado?
- O principal antigo é impedido de voltar aceitando gravações?
- A aplicação encontra o principal novo e reconecta sozinha?
Leituras
- Leituras na réplica são só as que aceitam atraso?
- Telas que mostram o que o usuário acabou de gravar usam "ler o que gravou", com o LSN lido depois da confirmação?
- Decisões de negócio leem do principal?
Monitoramento
- Há alerta para atraso de replicação, slots inativos e sessões esperando a réplica?
max_slot_wal_keep_sizeestá configurado?
Prompts para aprofundar
Ao abrir no ChatGPT ou no Claude, o prompt é enviado ao serviço escolhido pela URL e também fica copiado na sua área de transferência.
Montar uma réplica do PostgreSQL
Quero montar uma réplica de leitura do PostgreSQL para estudar. Pergunte se vou usar containers ou máquinas, e qual versão. Me guie, com os comandos completos: 1. O usuário de replicação, o pg_hba.conf e o slot. 2. O pg_basebackup com -R na réplica. 3. Conferir o estado da replicação nos dois lados. 4. Medir o atraso sob carga. 5. Promover a réplica e entender o que aconteceu com o principal antigo.
Implementar ler o que gravou
Quero implementar a técnica de ler o que gravou na minha aplicação, com réplicas de leitura. Pergunte qual linguagem, framework e driver eu uso, e como guardo a sessão do usuário. Vamos em etapas: 1. Ler o LSN depois do COMMIT de cada gravação. 2. Guardar o LSN da última gravação do usuário. 3. Antes de ler, conferir na réplica se ela já aplicou esse LSN, e cair para o principal se não. 4. Testes com a réplica pausada. 5. Métricas de quantas leituras caíram para o principal.
Revisar uma configuração de replicação
Vou colar abaixo a configuração de replicação do meu banco e o código que escolhe entre principal e réplica. Revise. Verifique, e para cada item diga se está correto, incorreto ou ausente, citando o trecho: 1. O modo de replicação combina com o que pode ser perdido? 2. Há réplica síncrona única que pode travar as gravações? 3. A troca de principal é automática e protegida contra dois principais? 4. Leituras críticas vão para o principal? 5. Telas que mostram o que o usuário gravou leem o que gravou? 6. Há limite de WAL retido por slot e monitoramento do atraso? Para cada problema, descreva o efeito numa falha e mostre a correção. Configuração e código: [cole aqui]