pkglog
EN

C++에서 PostgreSQL 연동: libpq vs libpqxx, 트랜잭션, Prepared Statement, JSONB와 연결 풀

작성 · · 30분 읽기 · 수정 고급 실습

들어가며: C++에서 PostgreSQL을 왜 쓰나요?

이 글이 답하는 질문

"주문·결제 데이터를 안전하게 저장해야 하는데, 어떻게 하죠?"
"사용자 입력을 그대로 쿼리에 넣으면 SQL injection 위험이 있습니다."
"DB 연결이 끊어지면 재연결·재시도 로직을 어떻게 구현하나요?"

PostgreSQL은 ACID를 보장하는 관계형 DB로, C++ 서버에서 주문·결제·재고, 복잡한 JOIN·트랜잭션, Prepared Statement를 통해 안전하고 효율적인 데이터 처리를 가능하게 합니다. 이 글은 libpq(C 기반, PostgreSQL 공식)와 libpqxx(Modern C++ 래퍼)로 PostgreSQL을 C++에서 연동하는 방법을, 설치부터 트랜잭션·연결 풀·재연결과 자주 만나는 에러까지 순서대로 다룹니다.

C++에서 DB를 다룰 때 가장 어려운 점은 SQL 자체보다 자원 수명과 에러 경로입니다. libpq는 C 라이브러리라 PGconn과 PGresult를 직접 해제해야 하고, 쿼리 실패는 반환 코드로만 알려 줍니다. 조건문 하나를 빠뜨리면 조용히 메모리가 새거나 실패한 쿼리 결과를 읽게 됩니다. 이 글의 예제가 대부분 RAII 래퍼와 예외를 중심으로 구성된 이유가 여기에 있습니다. 코드는 libpqxx 7.x API 기준이며, 8.x에서 이름이 바뀐 부분은 본문에서 따로 짚습니다.

요구 환경: C++17 이상(libpqxx 8.x는 C++20), PostgreSQL 12 이상 권장

일관성·SQL Injection·연결 풀: PostgreSQL 연동에서 겪는 상황

주문·결제 시 데이터 일관성

"주문 생성 후 결제 실패 시, 주문 상태를 어떻게 롤백하죠?"
"재고 차감과 주문 생성이 동시에 실패하면 안 돼요."

상황: 주문 생성과 결제 처리, 재고 차감이 여러 테이블에 걸쳐 있습니다. 중간에 실패하면 부분 커밋으로 데이터 불일치가 발생합니다. 해결 포인트: PostgreSQL 트랜잭션(BEGIN/COMMIT/ROLLBACK)으로 원자적 처리. libpqxx의 pqxx::work로 RAII 기반 자동 롤백 처리.

SQL Injection 방지

"사용자 입력을 그대로 쿼리에 넣으면 위험하다고 들었습니다."
"문자열 이스케이프를 어떻게 하죠?"

상황: "SELECT * FROM users WHERE name = '" + userInput + "'" 같은 문자열 연결은 SQL injection에 취약합니다. 해결 포인트: Prepared Statement로 파라미터 바인딩. $1, $2 플레이스홀더에 값을 바인딩하면 이스케이프가 자동 처리됩니다.

연결 풀 부족

"동시 요청이 많아지면 'too many connections' 에러가 나요."
"매 요청마다 새 연결을 만들면 느려요."

상황: 웹 서버가 요청마다 새 DB 연결을 생성하면, PostgreSQL max_connections 한도에 도달하고 연결 오버헤드로 지연이 발생합니다. 해결 포인트: 연결 풀로 연결 재사용. libpqxx 또는 PgBouncer와 함께 사용.

대용량 결과 처리

"10만 건 조회 시 메모리가 폭발해요."

상황: SELECT * FROM logs로 대량 조회 시 전체 결과를 메모리에 로드하면 OOM이 발생합니다. 해결 포인트: 커서(Cursor) 또는 스트리밍으로 청크 단위 처리. libpqxx의 stream_from 사용.

연결 끊김·재연결

"DB 서버 재시작 후 앱이 계속 에러를 내요."

상황: 장시간 유지 연결이 DB 재시작이나 네트워크로 끊기면, 이후 쿼리가 실패합니다. 해결 포인트: Health Check 및 재연결 로직. PQstatus() 체크 후 PQreset() 또는 새 연결 생성.

시나리오별 권장 패턴

시나리오해결책C++ 라이브러리
트랜잭션BEGIN/COMMIT/ROLLBACKlibpqxx::work
SQL injectionPrepared Statementlibpqxx::prepare
연결 풀Connection Poollibpqxx, PgBouncer
대용량 조회Cursor/스트리밍libpqxx::stream_from
재연결PQstatus + PQresetlibpq

PostgreSQL 서버와 libpq·libpqxx 설치

PostgreSQL 서버 실행

# Docker로 PostgreSQL 실행 (권장)
docker run -d -p 5432:5432 \
  -e POSTGRES_USER=postgres \
  -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=mydb \
  postgres:16-alpine
# 또는 로컬 설치 후
pg_ctl -D /usr/local/var/postgres start

libpq 설치

libpq는 PostgreSQL 공식 C 클라이언트 라이브러리입니다.

# Ubuntu/Debian
sudo apt-get install libpq-dev
# macOS (Homebrew)
brew install libpq
# vcpkg
vcpkg install libpq

libpqxx 설치

libpqxx는 libpq 위에 구축된 공식 C++ 래퍼입니다.

# vcpkg (권장)
vcpkg install libpqxx
# Ubuntu/Debian
sudo apt-get install libpqxx-dev
# macOS (Homebrew)
brew install libpqxx
# 또는 소스 빌드
git clone https://github.com/jtv/libpqxx.git
cd libpqxx
mkdir build && cd build
cmake .. -DCMAKE_BUILD_TYPE=Release
make && sudo make install

CMake 연동 예시

# CMakeLists.txt - libpq 사용
cmake_minimum_required(VERSION 3.16)
project(postgres_demo LANGUAGES CXX)
set(CMAKE_CXX_STANDARD 17)
find_package(PostgreSQL REQUIRED)
add_executable(pq_demo main.cpp)
target_link_libraries(pq_demo PRIVATE PostgreSQL::PostgreSQL)
# CMakeLists.txt - libpqxx 사용
cmake_minimum_required(VERSION 3.16)
project(postgres_demo LANGUAGES CXX)
set(CMAKE_CXX_STANDARD 17)
find_package(libpqxx REQUIRED)
add_executable(pqxx_demo main.cpp)
target_link_libraries(pqxx_demo PRIVATE libpqxx::pqxx)

설치에서 자주 막히는 부분은 배포판 패키지의 libpqxx 버전입니다. Ubuntu 22.04의 libpqxx-dev는 6.4 계열이라, 이 글의 7.x 문법(exec_prepared, stream_to::table, 구조화 바인딩 stream<>)이 컴파일되지 않습니다. no member named 'exec_prepared' 같은 에러가 나면 헤더 버전부터 확인하세요(grep PQXX_VERSION /usr/include/pqxx/version.hxx). 또 find_package(libpqxx)는 libpqxx가 CMake 설정 파일을 설치한 경우(vcpkg, 소스 빌드)에만 동작하고, 배포판 패키지로 설치했다면 pkg_check_modules(PQXX REQUIRED libpqxx)로 찾아야 할 때가 있습니다. macOS Homebrew의 libpq는 keg-only라서 /opt/homebrew/opt/libpq를 CMAKE_PREFIX_PATH에 넣어야 find_package(PostgreSQL)가 찾습니다.

libpq 연결과 파라미터 바인딩

아키텍처 다이어그램

flowchart TB
    subgraph App[C++ 애플리케이션]
        Main[main]
        Client[PgClient]
    end
    subgraph Libpq[libpq]
        Conn[PGconn]
        Result[PGresult]
        Exec[PQexec]
    end
    subgraph PG[PostgreSQL 서버]
        DB["(데이터베이스)"]
    end
    Main --> Client
    Client --> Conn
    Client --> Exec
    Exec --> Result
    Conn -->|TCP 5432| DB

연결 문자열 (Connection String)

postgresql://user:password@host:port/dbname
  • user: DB 사용자
  • password: 비밀번호
  • host: 호스트 (127.0.0.1 또는 로컬)
  • port: 포트 (기본 5432)
  • dbname: 데이터베이스 이름

기본 연결 (RAII)

// libpq_basic.cpp
// 컴파일: g++ -std=c++17 -o pq_basic libpq_basic.cpp -lpq
#include <libpq-fe.h>
#include <iostream>
#include <memory>
#include <stdexcept>
#include <string>
struct PgConnection {
    PGconn* conn = nullptr;
    PgConnection(const char* conninfo) {
        conn = PQconnectdb(conninfo);
        if (conn == nullptr) {
            throw std::runtime_error("PostgreSQL 연결 할당 실패");
        }
        if (PQstatus(conn) != CONNECTION_OK) {
            std::string err = PQerrorMessage(conn);
            PQfinish(conn);
            throw std::runtime_error("PostgreSQL 연결 실패: " + err);
        }
    }
    ~PgConnection() {
        if (conn) PQfinish(conn);
    }
    PgConnection(const PgConnection&) = delete;
    PgConnection& operator=(const PgConnection&) = delete;
};
int main() {
    try {
        PgConnection conn("host=127.0.0.1 port=5432 dbname=mydb user=postgres password=postgres");
        // 테이블 생성
        PGresult* res = PQexec(conn.conn, "CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY, name VARCHAR(100), email VARCHAR(100))");
        if (PQresultStatus(res) != PGRES_COMMAND_OK) {
            std::cerr << "CREATE TABLE 에러: " << PQerrorMessage(conn.conn) << "\n";
            PQclear(res);
            return 1;
        }
        PQclear(res);
        // INSERT
        const char* insertParams[] = {"홍길동", "[email protected]"};
        res = PQexecParams(conn.conn,
            "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id",
            2, nullptr, insertParams,
            nullptr, nullptr, 0);
        if (PQresultStatus(res) != PGRES_TUPLES_OK) {
            std::cerr << "INSERT 에러: " << PQerrorMessage(conn.conn) << "\n";
            PQclear(res);
            return 1;
        }
        std::cout << "INSERT 성공, id: " << PQgetvalue(res, 0, 0) << "\n";
        PQclear(res);
        // SELECT
        res = PQexec(conn.conn, "SELECT id, name, email FROM users");
        if (PQresultStatus(res) != PGRES_TUPLES_OK) {
            std::cerr << "SELECT 에러: " << PQerrorMessage(conn.conn) << "\n";
            PQclear(res);
            return 1;
        }
        int rows = PQntuples(res);
        for (int i = 0; i < rows; ++i) {
            std::cout << PQgetvalue(res, i, 0) << " | "
                      << PQgetvalue(res, i, 1) << " | "
                      << PQgetvalue(res, i, 2) << "\n";
        }
        PQclear(res);
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << "\n";
        return 1;
    }
    return 0;
}

이 코드에서 눈여겨볼 점은 결과 상태 코드가 쿼리 종류마다 다르다는 것입니다. CREATE TABLE처럼 행을 돌려주지 않는 명령은 PGRES_COMMAND_OK, SELECT나 RETURNING이 붙은 INSERT는 PGRES_TUPLES_OK가 성공입니다. 처음에는 RETURNING id를 붙인 INSERT를 PGRES_COMMAND_OK로 검사해서, 성공했는데도 에러 분기로 빠지는 실수를 하기 쉽습니다. 또 PQexec는 실패해도 대부분 nullptr이 아닌 결과 객체를 돌려주므로, 에러 분기에서도 PQclear를 반드시 호출해야 합니다. C 예제를 옮겨 오다 보면 (const char*[]){...} 형태의 복합 리터럴을 그대로 쓰게 되는데, 이는 C99 문법이라 표준 C++에서는 컴파일되지 않거나(taking address of temporary array) 컴파일러 확장에 의존하게 됩니다. 위처럼 이름 있는 배열로 선언하는 것이 안전합니다.

PQexecParams로 파라미터 바인딩 (SQL Injection 방지)

// PQexecParams: $1, $2 플레이스홀더에 값을 바인딩
// - SQL injection 방지 (값이 SQL 문과 분리되어 전송됨)
// - paramTypes가 nullptr이면 서버가 컬럼 문맥으로 타입 추론
// - 이름 없는 문장으로 매번 파싱되므로 플랜 재사용은 PQprepare로
const char* paramValues[] = {userName.c_str(), userEmail.c_str()};
const int paramLengths[] = {static_cast<int>(userName.size()), static_cast<int>(userEmail.size())};
const int paramFormats[] = {0, 0};  // 0 = 텍스트
PGresult* res = PQexecParams(conn,
    "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id",
    2, nullptr, paramValues, paramLengths, paramFormats, 0);

PGresult RAII 래퍼

// pg_result_guard.hpp
#pragma once
#include <libpq-fe.h>
#include <utility>
struct PgResultGuard {
    PGresult* res = nullptr;
    explicit PgResultGuard(PGresult* r) : res(r) {}
    ~PgResultGuard() { if (res) PQclear(res); }
    PgResultGuard(const PgResultGuard&) = delete;
    PgResultGuard& operator=(const PgResultGuard&) = delete;
    PgResultGuard(PgResultGuard&& other) noexcept : res(std::exchange(other.res, nullptr)) {}
    PGresult* get() const { return res; }
    PGresult* operator->() const { return res; }
};

PQexecParams가 SQL injection을 막는 원리는 이스케이프가 아니라 분리입니다. 쿼리 문자열과 파라미터 값이 프로토콜 수준에서 서로 다른 메시지(Parse와 Bind)로 전송되므로, 값 안에 따옴표나 세미콜론이 있어도 서버는 그것을 SQL 구문으로 해석하지 않습니다. 그래서 문자열을 직접 이스케이프하는 방식보다 구조적으로 안전합니다. 한 가지 제약은 플레이스홀더가 값 자리에만 쓸 수 있다는 점입니다. 테이블 이름, 컬럼 이름, ORDER BY 방향 같은 식별자는 $1로 바인딩할 수 없어서, 이런 부분이 동적이면 허용 목록(allowlist)과 대조하거나 PQescapeIdentifier로 감싸야 합니다.

PgResultGuard는 이동 생성자만 두고 이동 대입은 선언하지 않았으므로 이동 대입이 삭제된 상태입니다. 결과를 교체할 일이 있다면 reset() 같은 메서드를 추가하거나, 더 간단하게 std::unique_ptr<PGresult, decltype(&PQclear)>를 쓰는 방법도 있습니다.

libpqxx Modern C++ 클라이언트

연결 및 기본 사용

// libpqxx_basic.cpp
// vcpkg install libpqxx 후 컴파일
#include <pqxx/pqxx>
#include <iostream>
#include <string>
int main() {
    try {
        pqxx::connection conn("host=127.0.0.1 port=5432 dbname=mydb user=postgres password=postgres");
        // 테이블 생성
        pqxx::work w(conn);
        w.exec("CREATE TABLE IF NOT EXISTS products (id SERIAL PRIMARY KEY, name VARCHAR(100), price INTEGER)");
        w.commit();
        // INSERT
        pqxx::work w2(conn);
        w2.exec_params("INSERT INTO products (name, price) VALUES ($1, $2) RETURNING id",
                       "상품A", 9900);
        w2.commit();
        // SELECT
        pqxx::read_transaction r(conn);
        pqxx::result res = r.exec("SELECT id, name, price FROM products");
        for (auto row : res) {
            std::cout << row[0].as<int>() << " | "
                      << row[1].as<std::string>() << " | "
                      << row[2].as<int>() << "\n";
        }
    } catch (const pqxx::sql_error& e) {
        std::cerr << "SQL 에러: " << e.what() << "\n쿼리: " << e.query() << "\n";
        return 1;
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << "\n";
        return 1;
    }
    return 0;
}

트랜잭션 (RAII 자동 롤백)

// pqxx::work: 트랜잭션. commit() 호출 시 커밋, 예외 시 자동 롤백
pqxx::work w(conn);
try {
    w.exec_params("INSERT INTO orders (user_id, amount) VALUES ($1, $2)", 1, 10000);
    w.exec_params("UPDATE inventory SET stock = stock - 1 WHERE product_id = $1", 1);
    w.commit();  // 성공 시 커밋
} catch (...) {
    // w 소멸 시 자동 ROLLBACK
    throw;
}

catch (...)에서 throw;로 다시 던지는 것은 롤백을 위해서가 아닙니다. 롤백은 w의 소멸자가 알아서 하고, 다시 던지는 이유는 호출자에게 실패를 알리기 위해서입니다. 주의할 점은 트랜잭션 안에서 쿼리 하나가 실패하면 PostgreSQL이 그 트랜잭션 전체를 “aborted” 상태로 바꾼다는 것입니다. 이후 같은 트랜잭션에서 다른 쿼리를 보내면 current transaction is aborted, commands ignored until end of transaction block 에러만 돌아옵니다. 실패를 잡아서 계속 진행하고 싶다면 pqxx::subtransaction(SAVEPOINT)을 써야 합니다.

Prepared Statement

// Prepared Statement: 파싱 재사용, SQL injection 방지
// 연결마다 한 번만 prepare (같은 이름으로 두 번 prepare하면 에러)
conn.prepare("get_user", "SELECT id, name FROM users WHERE id = $1");
conn.prepare("insert_order", "INSERT INTO orders (user_id, amount) VALUES ($1, $2) RETURNING id");
pqxx::work w(conn);
pqxx::result r = w.exec_prepared("get_user", user_id);
pqxx::result r2 = w.exec_prepared("insert_order", user_id, amount);

인터넷의 오래된 예제에는 w.prepared("get_user")(user_id).exec() 같은 체이닝 문법이 많이 보이는데, 이는 libpqxx 6.x 이전 API로 7.0에서 제거되었습니다. 7.x에서는 exec_prepared(이름, 인자...)를 쓰고, 8.x에서는 w.exec(pqxx::prepped{"get_user"}, pqxx::params{user_id}) 형태로 다시 정리되었습니다. conn.prepare()는 서버에 즉시 PREPARE를 보내므로, 같은 이름으로 다시 호출하면 prepared statement "get_user" already exists 에러가 납니다. 그래서 prepare는 연결을 만든 직후 한 번만 하는 것이 원칙이고, 연결 풀을 쓴다면 풀이 새 연결을 만들 때 준비 문장을 함께 등록해야 합니다.

libpq vs libpqxx 비교

항목libpqlibpqxx
언어CC++17 (7.x) / C++20 (8.x)
의존성없음 (libpq만)libpq
트랜잭션수동 BEGIN/COMMITpqxx::work RAII
결과 처리PQgetvalue, 수동row.as()
PreparedPQprepareconn.prepare()
예외없음 (return code)예외 기반

CRUD 래퍼·주문 트랜잭션·스트리밍·연결 풀 예제

CRUD 래퍼 클래스 (libpqxx)

// user_repository.hpp
#pragma once
#include <pqxx/pqxx>
#include <optional>
#include <string>
#include <vector>
struct User {
    int id;
    std::string name;
    std::string email;
};
class UserRepository {
public:
    // prepare는 연결당 한 번만: 생성자에서 등록
    explicit UserRepository(pqxx::connection& conn) : conn_(conn) {
        conn_.prepare("find_user", "SELECT id, name, email FROM users WHERE id = $1");
        conn_.prepare("insert_user", "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id");
        conn_.prepare("update_user", "UPDATE users SET name = $1, email = $2 WHERE id = $3");
        conn_.prepare("delete_user", "DELETE FROM users WHERE id = $1");
    }
    std::optional<User> findById(int id) {
        pqxx::read_transaction r(conn_);
        auto res = r.exec_prepared("find_user", id);
        if (res.empty()) return std::nullopt;
        auto row = res[0];
        return User{
            row[0].as<int>(),
            row[1].as<std::string>(),
            row[2].as<std::string>()
        };
    }
    int insert(const std::string& name, const std::string& email) {
        pqxx::work w(conn_);
        auto res = w.exec_prepared("insert_user", name, email);
        int id = res[0][0].as<int>();
        w.commit();
        return id;
    }
    bool update(int id, const std::string& name, const std::string& email) {
        pqxx::work w(conn_);
        auto res = w.exec_prepared("update_user", name, email, id);
        bool ok = res.affected_rows() > 0;
        w.commit();
        return ok;
    }
    bool remove(int id) {
        pqxx::work w(conn_);
        auto res = w.exec_prepared("delete_user", id);
        bool ok = res.affected_rows() > 0;
        w.commit();
        return ok;
    }
private:
    pqxx::connection& conn_;
};

이 저장소 클래스는 연결을 참조로 받기 때문에 연결이 저장소보다 오래 살아야 하고, 하나의 연결을 쓰므로 여러 스레드에서 같은 저장소 객체를 동시에 쓰면 안 됩니다. 한 연결 위에서는 동시에 하나의 트랜잭션만 열 수 있어서, 메서드가 트랜잭션을 연 상태로 다른 메서드를 호출하면 Started new transaction while transaction was still active 같은 pqxx::usage_error가 납니다. 여러 작업을 한 트랜잭션으로 묶어야 한다면 메서드가 pqxx::work&를 인자로 받게 설계하는 편이 낫습니다. 또 findById는 결과가 없을 때 std::nullopt를 돌려주는데, email 컬럼이 NULL을 허용한다면 row[2].as<std::string>()이 pqxx::conversion_error를 던지므로 as<std::optional<std::string>>()로 받아야 합니다.

주문·결제 트랜잭션 (원자적 처리)

// order_service.cpp
#include <pqxx/pqxx>
#include <stdexcept>
#include <string>
struct OrderResult {
    int order_id;
    bool success;
    std::string error_message;
};
// 연결 생성 직후 한 번만 호출
void prepareOrderStatements(pqxx::connection& conn) {
    conn.prepare("insert_order", "INSERT INTO orders (user_id, product_id, quantity, amount) VALUES ($1, $2, $3, $4) RETURNING id");
    conn.prepare("update_inventory", "UPDATE inventory SET stock = stock - $1 WHERE product_id = $2 AND stock >= $1");
    conn.prepare("insert_payment", "INSERT INTO payments (order_id, amount) VALUES ($1, $2)");
}
OrderResult createOrderWithPayment(pqxx::connection& conn,
                                   int user_id,
                                   int product_id,
                                   int quantity,
                                   int amount) {
    pqxx::work w(conn);
    try {
        auto order_res = w.exec_prepared("insert_order", user_id, product_id, quantity, amount);
        int order_id = order_res[0][0].as<int>();
        auto inv_res = w.exec_prepared("update_inventory", quantity, product_id);
        if (inv_res.affected_rows() == 0) {
            throw std::runtime_error("재고 부족");
        }
        w.exec_prepared("insert_payment", order_id, amount);
        w.commit();
        return {order_id, true, ""};
    } catch (const std::exception& e) {
        // w 소멸 시 자동 ROLLBACK
        return {0, false, e.what()};
    }
}

재고 차감을 UPDATE ... SET stock = stock - $1 WHERE ... AND stock >= $1로 한 줄에 처리한 것이 이 예제의 핵심입니다. 흔히 SELECT stock으로 먼저 읽고 C++에서 비교한 뒤 UPDATE하는 코드를 쓰는데, 기본 격리 수준(READ COMMITTED)에서는 두 요청이 같은 재고를 동시에 읽고 둘 다 “재고 있음”으로 판단해 초과 판매가 생깁니다. 조건을 UPDATE의 WHERE에 넣으면 행 잠금을 잡은 상태에서 조건을 다시 평가하므로 이 경쟁 조건이 사라지고, affected_rows() == 0으로 재고 부족을 판정할 수 있습니다. 읽고 판단하는 로직이 복잡해서 한 문장으로 표현할 수 없다면 SELECT ... FOR UPDATE로 행을 먼저 잠그는 방식을 씁니다.

또 하나, 실제 결제 대행사(PG) 호출처럼 외부 네트워크 요청을 트랜잭션 안에서 하지 않는 것이 좋습니다. 외부 API가 몇 초씩 걸리면 그동안 행 잠금과 연결을 붙잡고 있어서 다른 주문이 줄줄이 대기합니다. 보통은 주문을 “결제 대기” 상태로 커밋하고, 외부 결제를 호출한 뒤 결과에 따라 별도 트랜잭션으로 상태를 갱신합니다.

대용량 데이터 스트리밍 (stream_to / stream)

// bulk_insert.cpp
// libpqxx stream_to: COPY 프로토콜로 대량 INSERT
#include <pqxx/pqxx>
#include <vector>
#include <string>
void bulkInsertProducts(pqxx::connection& conn,
                        const std::vector<std::pair<std::string, int>>& products) {
    pqxx::work w(conn);
    w.exec("CREATE TABLE IF NOT EXISTS products (id SERIAL PRIMARY KEY, name VARCHAR(100), price INTEGER)");
    auto stream = pqxx::stream_to::table(w, {"products"}, {"name", "price"});
    for (const auto& [name, price] : products) {
        stream.write_values(name, price);  // 한 행씩 기록
    }
    stream.complete();  // COPY 종료 (commit 전에 반드시 호출)
    w.commit();
}
// stream<>: SELECT 결과를 한 행씩 스트리밍으로 읽기 (내부적으로 COPY TO)
void streamLargeResult(pqxx::connection& conn) {
    pqxx::read_transaction r(conn);
    for (auto [id, name, price] : r.stream<int, std::string, int>(
            "SELECT id, name, price FROM products")) {
        // 10만 건이어도 한 번에 메모리에 로드하지 않음
        std::cout << id << " " << name << " " << price << "\n";
    }
}

이름이 헷갈리기 쉬운데, stream_to는 C++에서 DB로 쓰는 방향(COPY FROM STDIN), stream_from과 transaction::stream<>()은 DB에서 C++로 읽는 방향(COPY TO STDOUT)입니다. 스트림이 열려 있는 동안 그 연결은 COPY 모드라 다른 쿼리를 보낼 수 없으므로, 반복문 안에서 같은 트랜잭션으로 w.exec()를 호출하면 에러가 납니다. 또 complete()를 호출하지 않고 commit()하면 COPY가 끝나지 않은 상태라 예외가 발생합니다. 읽기 스트림에서 std::string으로 받은 컬럼이 NULL이면 변환 에러가 나므로, NULL 가능 컬럼은 std::optional<std::string>으로 받아야 합니다.

연결 풀 (간단한 구현)

// connection_pool.hpp
#pragma once
#include <pqxx/pqxx>
#include <condition_variable>
#include <mutex>
#include <queue>
#include <string>
class ConnectionPool {
public:
    ConnectionPool(const std::string& conninfo, size_t pool_size = 10)
        : conninfo_(conninfo) {
        for (size_t i = 0; i < pool_size; ++i) {
            pool_.push(std::make_unique<pqxx::connection>(conninfo_));
        }
    }
    std::unique_ptr<pqxx::connection> acquire() {
        std::unique_lock lock(mutex_);
        cv_.wait(lock, [this] { return !pool_.empty(); });
        auto conn = std::move(pool_.front());
        pool_.pop();
        return conn;
    }
    void release(std::unique_ptr<pqxx::connection> conn) {
        if (!conn) return;
        std::lock_guard lock(mutex_);
        pool_.push(std::move(conn));
        cv_.notify_one();
    }
private:
    std::string conninfo_;
    std::queue<std::unique_ptr<pqxx::connection>> pool_;
    std::mutex mutex_;
    std::condition_variable cv_;
};

이 풀은 개념을 보여 주기 위한 최소 구현이라 실제로 쓰기 전에 보완할 점이 있습니다. 첫째, acquire()와 release()를 짝지어 호출하는 책임이 사용자에게 있어서, 중간에 예외가 나면 연결이 풀로 돌아오지 않고 결국 모든 스레드가 cv_.wait에서 영원히 멈춥니다. 반환을 소멸자에서 처리하는 핸들 객체(커스텀 deleter를 가진 unique_ptr 등)로 감싸는 것이 안전합니다. 둘째, 무한 대기 대신 wait_for로 타임아웃을 두어야 DB가 느려졌을 때 요청이 쌓이지 않고 에러로 빠르게 실패합니다. 셋째, 풀에 돌려받은 연결이 끊겨 있을 수 있으므로 꺼낼 때 conn->is_open()을 확인하고 필요하면 새로 만들어야 합니다. 넷째, 트랜잭션을 커밋하지 않은 채 반환된 연결을 다음 사용자가 받으면 이상한 상태를 물려받으므로, 트랜잭션 객체를 연결보다 먼저 소멸시키는 순서를 지켜야 합니다.

재연결 로직 (libpq)

// reconnect.cpp
#include <libpq-fe.h>
#include <chrono>
#include <iostream>
#include <thread>
PGconn* ensureConnection(PGconn* conn, const char* conninfo) {
    if (conn && PQstatus(conn) == CONNECTION_OK) {
        return conn;
    }
    if (conn) {
        PQfinish(conn);
    }
    conn = PQconnectdb(conninfo);
    if (PQstatus(conn) != CONNECTION_OK) {
        std::cerr << "재연결 실패: " << PQerrorMessage(conn) << "\n";
        PQfinish(conn);
        return nullptr;
    }
    return conn;
}
// 또는 PQreset: 기존 연결 리소스 재사용
bool resetConnection(PGconn* conn) {
    if (PQstatus(conn) != CONNECTION_OK) {
        PQreset(conn);
        return PQstatus(conn) == CONNECTION_OK;
    }
    return true;
}

연결 거부, PQclear 누락, too many connections: 에러 해결

Connection refused / Connection timed out

증상: PQconnectdb 실패, PQerrorMessage에 “Connection refused” 또는 “Connection timed out” 원인:

  • PostgreSQL 서버가 실행 중이 아님
  • 잘못된 호스트/포트
  • 방화벽 차단
  • pg_hba.conf에서 클라이언트 IP 미허용 해결법:
// ❌ 잘못된 설정
PgConnection conn("host=wronghost port=5432 dbname=mydb");  // 잘못된 호스트
// ✅ 연결 문자열 검증
const char* conninfo = "host=127.0.0.1 port=5432 dbname=mydb user=postgres password=postgres connect_timeout=5";
PGconn* conn = PQconnectdb(conninfo);
if (PQstatus(conn) != CONNECTION_OK) {
    std::cerr << "연결 실패: " << PQerrorMessage(conn) << "\n";
    PQfinish(conn);
}
# PostgreSQL 서버 확인
pg_isready -h 127.0.0.1 -p 5432
# exit 0이면 정상

PQclear 누락으로 메모리 누수

증상: 장시간 실행 시 메모리 사용량이 계속 증가 원인: PQexec/PQexecParams가 반환하는 PGresult*를 PQclear로 해제하지 않음

// ❌ 메모리 누수
PGresult* res = PQexec(conn, "SELECT * FROM users");
// ... 사용 ...
// PQclear(res) 누락!

해결법:

// ✅ RAII 래퍼 사용
PgResultGuard guard(PQexec(conn, "SELECT * FROM users"));
PGresult* res = guard.get();
if (PQresultStatus(res) != PGRES_TUPLES_OK) {
    return;
}
// guard 소멸 시 자동 PQclear

SQL Injection

증상: 악의적 사용자 입력으로 인해 데이터 유출 또는 삭제 원인: 사용자 입력을 문자열 연결로 쿼리에 직접 삽입

// ❌ SQL injection 취약
std::string query = "SELECT * FROM users WHERE name = '" + userInput + "'";
PQexec(conn, query.c_str());
// userInput = "'; DROP TABLE users; --" 이면 테이블 삭제됨

해결법:

// ✅ PQexecParams 또는 Prepared Statement 사용
const char* paramValues[] = {userInput.c_str()};
PQexecParams(conn, "SELECT * FROM users WHERE name = $1", 1, nullptr, paramValues, nullptr, nullptr, 0);
// libpqxx
conn.prepare("get_user_by_name", "SELECT * FROM users WHERE name = $1");
r.exec_prepared("get_user_by_name", userInput);

too many connections

증상: FATAL: sorry, too many clients already 원인: PostgreSQL max_connections 한도 초과 (기본 100) 해결법:

  • 연결 풀 사용: 연결 재사용
  • PgBouncer 도입: 연결 풀링 프록시
  • max_connections 증가 (PostgreSQL 설정)
// ✅ 연결 풀 사용
ConnectionPool pool("host=127.0.0.1 dbname=mydb user=postgres password=postgres", 10);
auto conn = pool.acquire();
// ... 사용 ...
pool.release(std::move(conn));

max_connections를 무작정 올리는 것은 좋은 해결책이 아닙니다. PostgreSQL은 연결마다 별도의 백엔드 프로세스를 띄우고, 각 프로세스가 work_mem 등을 따로 쓰기 때문에 연결 수가 늘면 메모리 사용량과 컨텍스트 전환 비용이 함께 늘어납니다. 제가 보기에 흔한 실수는 서버 인스턴스를 늘리면서 인스턴스당 풀 크기를 그대로 두는 경우입니다. 인스턴스 10개 × 풀 20개면 200 연결이 되어 기본 한도를 넘습니다. 풀 크기는 “인스턴스 수 × 풀 크기 + 관리용 여유분”이 한도 안에 들어가도록 역산해서 정해야 합니다.

트랜잭션 중 연결 끊김

증상: PQexec 실패, “connection lost” 또는 “server closed the connection” 원인: 트랜잭션 진행 중 DB 재시작이나 네트워크 끊김 해결법:

// ✅ 재시도 전 PQreset 또는 새 연결
PGresult* res = PQexec(conn, "SELECT ...");
if (res == nullptr || PQresultStatus(res) == PGRES_FATAL_ERROR) {
    if (PQstatus(conn) != CONNECTION_OK) {
        PQreset(conn);
        if (PQstatus(conn) != CONNECTION_OK) {
            // 새 연결 생성 또는 에러 반환
        }
    }
    // 재시도
}

여기서 “재시도”에는 함정이 있습니다. 끊긴 연결에서 진행 중이던 트랜잭션은 서버 쪽에서 이미 롤백되었으므로, 재연결 후 마지막 쿼리 하나만 다시 보내면 트랜잭션 앞부분의 변경이 빠진 채 실행됩니다. 재시도는 항상 트랜잭션 전체 단위로 해야 합니다. 더 까다로운 경우는 COMMIT을 보낸 직후 연결이 끊긴 경우입니다. 서버가 커밋을 완료했는지 알 수 없으므로 무작정 재시도하면 주문이 두 번 들어갈 수 있습니다. 결제처럼 중복이 치명적인 작업은 요청마다 고유 키를 두고 UNIQUE 제약으로 중복 삽입을 막는(멱등성 키) 설계가 필요합니다.

NULL 값 처리

증상: PQgetvalue가 NULL 반환 시 std::stoi 등에서 크래시 원인: DB 컬럼이 NULL일 수 있는데 NULL 체크 없이 사용

// ❌ NULL 미처리
int id = std::stoi(PQgetvalue(res, 0, 0));  // NULL이면 "NULL" 문자열이 아님, PQgetisnull 확인 필요
// ✅ NULL 체크
if (PQgetisnull(res, 0, 0)) {
    // NULL 처리
} else {
    int id = std::stoi(PQgetvalue(res, 0, 0));
}
// libpqxx: row[0].is_null() 체크
if (row[0].is_null()) {
    // NULL 처리
} else {
    int id = row[0].as<int>();
}

동일 연결을 멀티스레드에서 공유

증상: 간헐적 크래시, 잘못된 결과 원인: libpq PGconn은 스레드 안전하지 않음

// ❌ 위험
PGconn* conn = PQconnectdb(conninfo);
std::thread t1([&]() { PQexec(conn, "SELECT 1"); });
std::thread t2([&]() { PQexec(conn, "SELECT 2"); });

해결법:

// ✅ 스레드당 연결 또는 연결 풀
void worker() {
    thread_local pqxx::connection conn(conninfo);
    pqxx::work w(conn);
    w.exec("SELECT ...");
}

Prepared Statement·COPY·스트리밍으로 성능 올리기

Prepared Statement 사용

동일 쿼리를 반복 실행할 때 Prepared Statement로 쿼리 플랜 재사용. 파싱·플랜 최적화 비용을 줄입니다.

// ❌ 매번 파싱
for (int i = 0; i < 1000; ++i) {
    w.exec_params("SELECT * FROM users WHERE id = $1", i);
}
// ✅ Prepared Statement
conn.prepare("get_user", "SELECT * FROM users WHERE id = $1");
for (int i = 0; i < 1000; ++i) {
    w.exec_prepared("get_user", i);
}

다만 이 예제처럼 1000번을 반복한다면 병목은 파싱보다 1000번의 네트워크 왕복인 경우가 많습니다. 로컬 DB에서는 차이가 작아 보여도, DB가 다른 가용 영역에 있어 왕복이 1ms만 되어도 1000번이면 1초가 됩니다. 가능하면 WHERE id = ANY($1)에 배열을 넘기거나 JOIN으로 한 번에 가져오는 방식이 효과가 더 큽니다.

COPY로 대량 INSERT

행마다 INSERT를 보내는 대신 COPY로 데이터를 한 번에 흘려보내면 문장 처리와 왕복 비용이 크게 줄어듭니다.

// libpqxx stream_to
pqxx::work w(conn);
auto stream = pqxx::stream_to::table(w, {"products"}, {"name", "price"});
for (const auto& [name, price] : products) {
    stream.write_values(name, price);
}
stream.complete();
w.commit();

연결 풀 사용

매 요청마다 새 연결을 만들면 TCP 핸드셰이크·인증 비용이 큽니다. 연결 풀로 재사용하세요.

// libpqxx: ConnectionPool 사용 또는 PgBouncer

대용량 결과는 스트리밍

10만 건 이상 조회 시 pqxx::stream 또는 stream_from 사용

// ❌ 전체 메모리 로드
pqxx::result res = r.exec("SELECT * FROM large_table");
// 10만 건 * 1KB = 100MB 이상
// ✅ 스트리밍
for (auto [id, name] : r.stream<int, std::string>("SELECT id, name FROM large_table")) {
    process(id, name);
}

인덱스 활용

-- 자주 조회하는 컬럼에 인덱스
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_orders_user_id ON orders(user_id);

배치 커밋

대량 INSERT 시 INSERT ... VALUES (...), (...), (...) 여러 행을 한 번에 처리

// 1000건을 100건씩 묶어서 INSERT
std::string values;
for (size_t i = 0; i < batch.size(); i += 100) {
    values.clear();
    for (size_t j = i; j < std::min(i + 100, batch.size()); ++j) {
        if (j > i) values += ",";
        // w.quote()는 따옴표로 감싸고 이스케이프까지 해 줌 (직접 '...' 연결 금지)
        values += "(" + w.quote(batch[j].name) + "," + std::to_string(batch[j].price) + ")";
    }
    w.exec("INSERT INTO products (name, price) VALUES " + values);
}
w.commit();

재연결·재시도·트랜잭션 격리 수준

Health Check 및 재연결

bool isConnectionHealthy(pqxx::connection& conn) {
    try {
        pqxx::nontransaction n(conn);
        n.exec("SELECT 1");
        return true;
    } catch (...) {
        return false;
    }
}
void ensureConnection(pqxx::connection& conn, const std::string& conninfo) {
    if (!isConnectionHealthy(conn)) {
        conn.close();
        conn = pqxx::connection(conninfo);
    }
}

설정 외부화

struct PgConfig {
    std::string host = "127.0.0.1";
    int port = 5432;
    std::string dbname = "mydb";
    std::string user = "postgres";
    std::string password;
};
PgConfig loadFromEnv() {
    PgConfig c;
    if (const char* h = std::getenv("PGHOST")) c.host = h;
    if (const char* p = std::getenv("PGPORT")) c.port = std::stoi(p);
    if (const char* d = std::getenv("PGDATABASE")) c.dbname = d;
    if (const char* u = std::getenv("PGUSER")) c.user = u;
    if (const char* pw = std::getenv("PGPASSWORD")) c.password = pw;
    return c;
}
std::string toConnectionString(const PgConfig& c) {
    return "host=" + c.host + " port=" + std::to_string(c.port) +
           " dbname=" + c.dbname + " user=" + c.user +
           " password=" + c.password;
}

재시도 로직 (지수 백오프)

template <typename Func>
auto retryWithBackoff(Func&& f, int max_retries = 3) {
    for (int i = 0; i < max_retries; ++i) {
        try {
            return f();
        } catch (const pqxx::serialization_failure&) {  // 격리 수준 충돌: 재시도 가치 있음
            if (i == max_retries - 1) throw;
            std::this_thread::sleep_for(std::chrono::milliseconds(100 * (1 << i)));
        } catch (const pqxx::broken_connection&) {  // 연결 끊김: 재연결 후 재시도
            if (i == max_retries - 1) throw;
            std::this_thread::sleep_for(std::chrono::milliseconds(100 * (1 << i)));
        }
    }
    throw std::runtime_error("재시도 실패");
}
// 사용
retryWithBackoff([&]() {
    pqxx::work w(conn);
    w.exec("INSERT INTO ...");
    w.commit();
});

재시도 대상 예외를 좁힌 데는 이유가 있습니다. pqxx::sql_error 전체를 잡으면 문법 오류나 제약 조건 위반처럼 몇 번을 다시 해도 똑같이 실패할 에러까지 재시도하게 되어, 실패 응답만 늦어지고 로그가 세 배로 늘어납니다. 재시도할 가치가 있는 것은 직렬화 실패(SQLSTATE 40001), 데드락 감지(40P01, libpqxx의 deadlock_detected), 그리고 연결 끊김 정도입니다. broken_connection은 sql_error의 하위 클래스가 아니라 별도로 잡아야 한다는 점도 주의해야 합니다. 그리고 람다 안에서 매번 새 pqxx::work를 만드는 구조여야 트랜잭션 전체가 재시도됩니다.

로깅 및 모니터링

class LoggingConnection {
public:
    explicit LoggingConnection(const std::string& conninfo) : conn_(conninfo) {}
    pqxx::result exec(const std::string& query) {
        auto start = std::chrono::steady_clock::now();
        pqxx::nontransaction n(conn_);
        auto res = n.exec(query);
        auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now() - start);
        log("Query: " + query + " elapsed: " + std::to_string(elapsed.count()) + "ms");
        return res;
    }
private:
    pqxx::connection conn_;
};

steady_clock::now() 두 값의 차는 구현에 따라 나노초 단위 duration이므로, count()를 그대로 찍고 “ms”를 붙이면 값이 백만 배 부풀려 보입니다. 위처럼 duration_cast로 단위를 명시해야 합니다. 쿼리 전문을 로그에 남길 때는 파라미터에 개인정보나 비밀번호가 섞이지 않게 쿼리 텍스트만 기록하고, 서버 쪽에서는 log_min_duration_statement나 pg_stat_statements 확장으로 느린 쿼리를 모으는 편이 더 정확합니다.

트랜잭션 격리 수준

// READ COMMITTED (기본) | REPEATABLE READ | SERIALIZABLE
// 격리 수준은 트랜잭션 타입의 템플릿 인자로 지정 (BEGIN 시점에 적용)
pqxx::transaction<pqxx::isolation_level::repeatable_read> w(conn);
auto res = w.exec("SELECT stock FROM inventory WHERE product_id = 1");
w.commit();

SET TRANSACTION ISOLATION LEVEL은 트랜잭션 안에서 첫 쿼리보다 먼저 실행되어야 하고 그 트랜잭션에만 적용됩니다. pqxx::work를 연 뒤 이 문장만 실행하고 커밋하면 아무 효과도 없는 빈 트랜잭션이 됩니다. libpqxx에서는 위처럼 트랜잭션 타입에 격리 수준을 지정하는 것이 정확한 방법입니다. REPEATABLE READ 이상에서는 동시 수정이 충돌하면 could not serialize access due to concurrent update 에러가 나는데, 이는 버그가 아니라 “처음부터 다시 하라”는 신호이므로 앞의 재시도 로직과 함께 써야 합니다.

PostgreSQL 연동 점검 항목

환경 설정

  • PostgreSQL 서버 실행 확인 (pg_isready)
  • libpq 또는 libpqxx 설치
  • CMake/vcpkg 연동

연결 및 기본 사용

  • PQconnectdb 또는 pqxx::connection으로 연결
  • RAII로 PGconn/PGresult 관리
  • PQclear 누락 없이 호출 (libpq)

에러 처리

  • PQstatus(conn) != CONNECTION_OK 체크
  • PQresultStatus(res) 체크
  • pqxx::sql_error 예외 처리

보안

  • Prepared Statement 또는 PQexecParams 사용 (SQL injection 방지)
  • 비밀번호 환경 변수 사용

성능

  • 연결 풀 또는 스레드당 연결
  • Prepared Statement로 반복 쿼리 최적화
  • 대용량 조회 시 스트리밍

프로덕션

  • Health Check 주기적 수행, 재연결·재시도 정책

라이브러리·기능별 요약

항목libpqlibpqxx
용도C 호환, 경량, 임베디드Modern C++, 풍부한 API
연결PGconn 직접 관리pqxx::connection
트랜잭션수동 BEGIN/COMMITpqxx::work RAII
에러return code예외 기반
권장레거시, 최소 의존성신규 프로젝트

핵심 원칙:

  1. RAII로 연결·결과 관리
  2. Prepared Statement로 SQL injection 방지
  3. 멀티스레드에서는 연결 풀 또는 스레드당 연결
  4. 트랜잭션은 pqxx::work로 자동 롤백 보장 다음 글 Redis C++(#52-2)에서는 캐싱, 세션, 분산락을 다룹니다.

자주 묻는 질문 (FAQ)

Q. PostgreSQL의 NULL 값을 C++에서 빈 문자열과 어떻게 구분하나요?

A. libpq의 PQgetvalue는 값이 NULL이어도 빈 문자열을 돌려주기 때문에, 반환값만 보면 NULL과 실제 빈 문자열을 구분할 수 없습니다. libpq에서는 PQgetisnull로 먼저 NULL 여부를 확인해야 하고, libpqxx에서는 field의 is_null()을 확인하거나 std::optional로 변환해 받는 방식을 씁니다. NULL일 수 있는 컬럼을 바로 정수 등으로 변환하면 예외가 나거나 잘못된 값이 들어가므로, 스키마에서 NULL 허용 여부를 확인하고 그에 맞는 타입으로 받는 것이 좋습니다.

참고 자료

같이 보면 좋은 글