使用tokio-postgres执行START_REPLICATION命令失败，提示语法错误

你遇到的问题核心在于——START_REPLICATION不是普通的SQL查询命令，它属于PostgreSQL的复制协议专属指令，不能用simple_query这类执行常规SQL的方法来调用。当你用simple_query发送这个命令时，PostgreSQL会以常规SQL解析器去处理它，自然会抛出语法错误。

下面是具体的解决思路和修改后的代码示例：

解决步骤 #

1. 调整数据库连接参数 #

建立连接时，需要明确指定复制模式，在连接字符串中添加replication=database参数，这样PostgreSQL才会以复制连接的模式处理这个会话：

pub async fn logical_replication_connection() -> Result<tokio_postgres::Client, Box<dyn std::error::Error>> {
    println!("Consuming replication...");
    // 添加replication=database参数，启用复制连接模式
    let conn_str = "host=localhost user=postgres password=postgres dbname=mydb replication=database";
    let (client, connection) = tokio_postgres::connect(conn_str, NoTls).await?;
    
    tokio::spawn(async move {
        if let Err(e) = connection.await {
            eprintln!("connection error: {}", e);
        }
    });
    
    Ok(client)
}

2. 使用复制专用API启动逻辑复制 #

tokio-postgres提供了copy_both方法来处理双向复制流（逻辑复制需要发送确认和接收数据，属于双向流），替代原来的simple_query：

use tokio_postgres::replication::ReplicationStream;
use tokio_postgres::types::ToSql;

async fn process_replication(client: tokio_postgres::Client) {
    println!("Processing replication....");
    
    // 构造START_REPLICATION的核心参数
    let slot_name = "my_slot";
    let start_lsn = 0u64;
    let options = [
        ("proto_version", &1i32 as &dyn ToSql),
        ("publication_names", &"my_pub" as &dyn ToSql),
    ];
    
    // 使用copy_both启动逻辑复制，获取复制流
    match client.copy_both::<_, _, _>(
        format!("START_REPLICATION SLOT {} LOGICAL {}", slot_name, start_lsn),
        options.iter().map(|(k, v)| (*k, *v))
    ).await {
        Ok(stream) => {
            let mut replication_stream = ReplicationStream::new(stream);
            
            // 循环读取复制消息
            while let Some(msg) = replication_stream.next().await {
                match msg {
                    Ok(message) => {
                        match message {
                            tokio_postgres::replication::ReplicationMessage::XLogData(xlog_data) => {
                                // 这里是WAL原始数据，需要用pgoutput等插件解析为业务可读的变更
                                println!("Received WAL segment data (length: {})", xlog_data.data().len());
                            }
                            tokio_postgres::replication::ReplicationMessage::KeepAlive(keepalive) => {
                                // 回复心跳确认，维持复制连接
                                if let Err(e) = replication_stream.standby_status_update(
                                    keepalive.wal_end(),
                                    keepalive.wal_end(),
                                    0,
                                    true
                                ).await {
                                    eprintln!("Failed to send keepalive response: {}", e);
                                }
                            }
                            _ => {}
                        }
                    }
                    Err(e) => {
                        eprintln!("Replication stream error: {}", e);
                        break;
                    }
                }
            }
        }
        Err(e) => eprintln!("Failed to start replication: {}", e),
    }
}

额外说明 #

• 逻辑复制返回的是WAL原始二进制数据，你需要借助PostgreSQL内置的pgoutput协议（或第三方解码插件）来解析成具体的表变更数据，也可以使用tokio-postgres-replication这类封装好的库来简化解析工作。
• 确保你的数据库用户拥有REPLICATION权限，且复制槽my_slot、发布my_pub已经提前正确创建。

内容来源于stack exchange