C++에서 Redis 고급 기능 쓰기: Pub/Sub, 파이프라인 RTT 절감, Lua 스크립트, 트랜잭션, 클러스터

들어가며: “실시간 알림·대량 처리·원자적 작업이 막막해요”

실시간 알림·대량 처리·원자적 작업에서 막히는 상황

52-2에서 Redis 기본 연결·GET/SET·캐싱·분산 락을 다뤘다면, 이 글에서는 고급 기능을 다룹니다. 실무에서 자주 맞닥뜨리는 문제와 해결 방법을 제시합니다.

시나리오 1: 실시간 알림·채팅에서 서버가 클라이언트에게 푸시

상황: 주문 완료·채팅 메시지 등 이벤트 발생 시 연결된 모든 클라이언트가 즉시 알림을 받아야 함
문제: 단순 폴링은 지연·부하가 크고, WebSocket만으로는 다중 서버 간 메시지 공유 불가
결과: Redis Pub/Sub으로 메시지 브로드캐스트 → 여러 서버가 같은 채널 구독·발행

시나리오 2: 1000개 키를 한 번에 SET/GET할 때 RTT 폭증

상황: 캐시 워밍업·대량 조회 시 명령당 1ms RTT라면 1000건 = 1초 지연
문제: 단일 명령 반복으로 네트워크 왕복이 누적됨
결과: 파이프라인(Pipeline)으로 여러 명령을 한 번에 전송 → RTT 1회로 감소

시나리오 3: 재고 차감·분산 락 해제 시 경쟁 조건

상황: GET → 값 감소 → SET은 원자적이지 않아, 동시 요청 시 재고가 음수로 떨어질 수 있음
문제: 락 해제 시 "같은 토큰일 때만 DEL"을 클라이언트에서 여러 명령으로 처리하면 경쟁 조건
결과: Lua 스크립트로 서버에서 원자적으로 실행 → EVAL/EVALSHA

시나리오 4: 여러 키에 걸친 원자적 작업

상황: 주문 생성 시 재고 차감 + 주문 기록 + 포인트 적립을 한 번에 처리해야 함
문제: 중간에 실패하면 일부만 반영되어 데이터 불일치
결과: MULTI/EXEC 트랜잭션 또는 Lua 스크립트로 원자적 실행

시나리오 5: Redis Cluster에서 MOVED/ASK 에러

상황: 단일 연결로 Cluster에 접속 시 "MOVED 1234 127.0.0.1:7001" 에러
문제: hiredis 단일 연결은 슬롯 리다이렉트를 자동 처리하지 않음
결과: redis-plus-plus RedisCluster 또는 hiredis Cluster API 사용

flowchart TB
    subgraph 문제[실무 문제]
        P1[실시간 알림] --> S1[Pub/Sub]
        P2[대량 RTT] --> S2[파이프라인]
        P3[경쟁 조건] --> S3[Lua 스크립트]
        P4[원자적 작업] --> S4[트랜잭션/Lua]
        P5[Cluster 분산] --> S5[RedisCluster]
    end

요구 환경: C++17 이상, Redis 6.x 이상, hiredis·redis-plus-plus

---

Pub/Sub으로 실시간 메시징

Pub/Sub이란?

발행자(Publisher)가 채널에 메시지를 보내면, 구독자(Subscriber)가 해당 채널을 구독 중인 모든 연결에 메시지가 전달됩니다. 1:N, N:N 브로드캐스트에 적합합니다.

sequenceDiagram
    participant P as Publisher
    participant R as Redis
    participant S1 as Subscriber 1
    participant S2 as Subscriber 2
    S1->>R: SUBSCRIBE channel:orders
    S2->>R: SUBSCRIBE channel:orders
    P->>R: PUBLISH channel:orders "주문 완료 #123"
    R->>S1: message
    R->>S2: message

hiredis로 Pub/Sub

hiredis는 redisCommand를 사용하면 블로킹됩니다. SUBSCRIBE 후에는 다른 명령을 보낼 수 없으므로 별도 연결(또는 스레드)에서 구독해야 합니다.

// pubsub_hiredis.cpp
// 컴파일: g++ -std=c++17 -o pubsub pubsub_hiredis.cpp -lhiredis -lpthread
#include <hiredis/hiredis.h>
#include <iostream>
#include <memory>
#include <stdexcept>
#include <string>
#include <thread>
#include <chrono>
struct RedisConnection {
    redisContext* ctx = nullptr;
    RedisConnection(const char* host, int port) {
        ctx = redisConnect(host, port);
        if (!ctx || ctx->err) {
            throw std::runtime_error(ctx ? ctx->errstr : "연결 실패");
        }
    }
    ~RedisConnection() { if (ctx) redisFree(ctx); }
};
// 발행자: 별도 연결로 PUBLISH
void publisher(const std::string& channel) {
    RedisConnection conn("127.0.0.1", 6379);
    for (int i = 0; i < 5; ++i) {
        redisReply* r = (redisReply*)redisCommand(conn.ctx, "PUBLISH %s %s",
            channel.c_str(), ("메시지 " + std::to_string(i)).c_str());
        if (r) {
            std::cout << "[Publish] 수신자 수: " << r->integer << "\n";
            freeReplyObject(r);
        }
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}
// 구독자: SUBSCRIBE는 블로킹이므로 별도 스레드
void subscriber(const std::string& channel) {
    RedisConnection conn("127.0.0.1", 6379);
    redisReply* r = (redisReply*)redisCommand(conn.ctx, "SUBSCRIBE %s", channel.c_str());
    freeReplyObject(r);
    while (true) {
        if (redisGetReply(conn.ctx, (void**)&r) != REDIS_OK) break;
        // "message" 응답은 [종류, 채널, 내용] 3개 원소. 구독 확인 응답은 3번째가 정수라 str이 없음
        if (r->type == REDIS_REPLY_ARRAY && r->elements == 3 &&
            r->element[2]->type == REDIS_REPLY_STRING) {
            std::string type = r->element[0]->str;
            std::string msg(r->element[2]->str, r->element[2]->len);
            std::cout << "[Sub] " << type << ": " << msg << "\n";
        }
        freeReplyObject(r);
    }
}
int main() {
    std::string channel = "channel:orders";
    std::thread sub(subscriber, channel);
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    std::thread pub(publisher, channel);
    pub.join();
    sub.join();  // Ctrl+C로 종료 전까지 대기
    return 0;
}

주의: SUBSCRIBE 이후 해당 연결은 구독 모드가 되어 GET, SET 등 일반 명령을 사용할 수 없습니다. 발행·구독은 독립된 연결로 처리해야 합니다.

redis-plus-plus로 Pub/Sub

redis-plus-plus는 subscriber()로 구독 전용 연결을 만들고, on_message 콜백으로 메시지를 수신합니다. consume()은 응답 하나를 읽어 콜백을 호출하고 반환하므로, 계속 받으려면 루프 안에서 반복 호출해야 합니다.

// pubsub_redispp.cpp
// vcpkg install redis-plus-plus
#include <sw/redis++/redis++.h>
#include <iostream>
#include <thread>
#include <chrono>
using namespace sw::redis;
void run_publisher(const std::string& channel) {
    auto redis = Redis("tcp://127.0.0.1:6379");
    for (int i = 0; i < 5; ++i) {
        auto count = redis.publish(channel, "주문 #" + std::to_string(i + 100));
        std::cout << "[Publish] 수신자: " << count << "\n";
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}
void run_subscriber(const std::string& channel) {
    auto sub = Redis("tcp://127.0.0.1:6379").subscriber();
    sub.on_message([channel](std::string ch, std::string msg) {
        std::cout << "[Sub] " << ch << ": " << msg << "\n";
    });
    sub.subscribe(channel);
    while (true) {
        sub.consume();  // 메시지 하나를 받을 때까지 블로킹
    }
}
int main() {
    std::string channel = "channel:orders";
    std::thread sub(run_subscriber, channel);
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    std::thread pub(run_publisher, channel);
    pub.join();
    sub.join();
    return 0;
}

패턴 구독 (PSUBSCRIBE)

channel:orders:*처럼 와일드카드로 여러 채널을 한 번에 구독할 수 있습니다.

// hiredis PSUBSCRIBE
redisReply* r = (redisReply*)redisCommand(ctx, "PSUBSCRIBE channel:orders:*");
freeReplyObject(r);
// redis-plus-plus
sub.psubscribe("channel:orders:*");

---

파이프라인으로 RTT 줄이기

파이프라인이란?

여러 명령을 응답을 기다리지 않고 연달아 보낸 뒤, 응답을 순서대로 받습니다. 명령 N개에 왕복 N번이 들던 것이 사실상 왕복 1번 수준으로 줄어듭니다. 명령당 RTT가 1ms라면 1000개를 하나씩 보낼 때 왕복 대기만 1초인데, 파이프라인으로 보내면 그 대기가 거의 사라집니다. 원자성은 보장되지 않으므로, 중간에 다른 클라이언트의 명령이 끼어들 수 있습니다.

flowchart LR
    subgraph 단일["단일 명령 (RTT 3회)"]
        A1[Req1] --> A2[Res1]
        A2 --> B1[Req2]
        B1 --> B2[Res2]
        B2 --> C1[Req3]
        C1 --> C2[Res3]
    end
    subgraph 파이프["파이프라인 (RTT 1회)"]
        P1[Req1+Req2+Req3] --> P2[Res1+Res2+Res3]
    end

hiredis 파이프라인

redisAppendCommand로 명령을 버퍼에 추가하며, redisGetReply로 응답을 순서대로 받습니다.

// pipeline_hiredis.cpp
// 컴파일: g++ -std=c++17 -o pipeline pipeline_hiredis.cpp -lhiredis
#include <hiredis/hiredis.h>
#include <chrono>
#include <iostream>
#include <stdexcept>
#include <vector>
#include <string>
struct RedisConnection {
    redisContext* ctx = nullptr;
    RedisConnection(const char* host, int port) {
        ctx = redisConnect(host, port);
        if (!ctx || ctx->err) throw std::runtime_error("Redis 연결 실패");
    }
    ~RedisConnection() { if (ctx) redisFree(ctx); }
};
void pipeline_batch_set(redisContext* ctx, int count) {
    for (int i = 0; i < count; ++i) {
        std::string key = "key:" + std::to_string(i);
        std::string val = "value:" + std::to_string(i);
        redisAppendCommand(ctx, "SET %s %s", key.c_str(), val.c_str());
    }
    redisReply* reply = nullptr;
    for (int i = 0; i < count; ++i) {
        if (redisGetReply(ctx, (void**)&reply) != REDIS_OK) {
            std::cerr << "GetReply 실패\n";
            break;
        }
        freeReplyObject(reply);
    }
}
void pipeline_batch_get(redisContext* ctx, int count) {
    for (int i = 0; i < count; ++i) {
        std::string key = "key:" + std::to_string(i);
        redisAppendCommand(ctx, "GET %s", key.c_str());
    }
    redisReply* reply = nullptr;
    for (int i = 0; i < count; ++i) {
        if (redisGetReply(ctx, (void**)&reply) != REDIS_OK) break;
        if (reply->type == REDIS_REPLY_STRING) {
            // std::cout << reply->str << "\n";
        }
        freeReplyObject(reply);
    }
}
int main() {
    RedisConnection conn("127.0.0.1", 6379);
    auto start = std::chrono::high_resolution_clock::now();
    pipeline_batch_set(conn.ctx, 1000);
    pipeline_batch_get(conn.ctx, 1000);
    auto end = std::chrono::high_resolution_clock::now();
    auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
    std::cout << "파이프라인 2000건: " << ms << " ms\n";
    return 0;
}

주의: redisAppendCommand와 redisGetReply 사이에 다른 명령을 보내면 파이프라인 순서가 깨집니다. 한 번에 묶어서 처리하세요.

redis-plus-plus 파이프라인

redis.pipeline()으로 Pipeline 객체를 만들고, command()로 명령을 추가한 뒤 exec()로 실행합니다.

// pipeline_redispp.cpp
#include <sw/redis++/redis++.h>
#include <iostream>
#include <chrono>
using namespace sw::redis;
int main() {
    auto redis = Redis("tcp://127.0.0.1:6379");
    auto pipe = redis.pipeline();
    // 1000개 SET
    for (int i = 0; i < 1000; ++i) {
        pipe.set("key:" + std::to_string(i), "value:" + std::to_string(i));
    }
    auto replies = pipe.exec();
    std::cout << "파이프라인 SET 완료: " << replies.size() << " 건\n";
    // 1000개 GET
    auto pipe2 = redis.pipeline();
    std::vector<std::string> keys;
    for (int i = 0; i < 1000; ++i) {
        keys.push_back("key:" + std::to_string(i));
    }
    for (const auto& k : keys) {
        pipe2.get(k);
    }
    auto get_replies = pipe2.exec();
    std::cout << "파이프라인 GET 완료: " << get_replies.size() << " 건\n";
    return 0;
}

주의: redis-plus-plus Pipeline은 스레드 안전하지 않습니다. 멀티스레드에서 사용 시 std::mutex로 보호하세요.

---

Lua 스크립팅과 EVALSHA

Lua가 필요한 이유

Redis 명령은 각각 원자적이지만, 여러 명령을 묶어서 원자적으로 실행하려면 Lua가 필요합니다. Lua 스크립트는 Redis 서버에서 한 번에 실행되므로 네트워크 왕복 없이 원자성이 보장됩니다.

flowchart TB
    subgraph 클라이언트["클라이언트 (경쟁 조건)"]
        C1[GET stock] --> C2[감소]
        C2 --> C3[SET stock]
        C3 --> C4[다른 요청이 동시에 GET]
    end
    subgraph Lua["Lua (원자적)"]
        L1[EVAL 스크립트] --> L2[GET → 감소 → SET 한 번에]
    end

Lua 스크립트 기본

-- decrement_stock.lua
-- KEYS[1]: 재고 키
-- ARGV[1]: 차감 수량
local current = redis.call('GET', KEYS[1])
if not current then
    return -1  -- 키 없음
end
local stock = tonumber(current)
local amount = tonumber(ARGV[1])
if stock < amount then
    return -2  -- 재고 부족
end
redis.call('SET', KEYS[1], stock - amount)
return stock - amount

hiredis로 EVAL

// lua_hiredis.cpp
#include <hiredis/hiredis.h>
#include <iostream>
#include <string>
const char* DECREMENT_STOCK = R"(
local current = redis.call('GET', KEYS[1])
if not current then return -1 end
local stock = tonumber(current)
local amount = tonumber(ARGV[1])
if stock < amount then return -2 end
redis.call('SET', KEYS[1], stock - amount)
return stock - amount
)";
int main() {
    redisContext* ctx = redisConnect("127.0.0.1", 6379);
    if (!ctx || ctx->err) {
        std::cerr << "연결 실패\n";
        return 1;
    }
    // SET stock 100
    redisReply* r = (redisReply*)redisCommand(ctx, "SET stock 100");
    freeReplyObject(r);
    // EVAL decrement_stock
    r = (redisReply*)redisCommand(ctx,
        "EVAL %s 1 stock 5",
        DECREMENT_STOCK);
    if (r && r->type == REDIS_REPLY_INTEGER) {
        std::cout << "차감 후 재고: " << r->integer << "\n";  // 95
    }
    freeReplyObject(r);
    redisFree(ctx);
    return 0;
}

redis-plus-plus로 EVAL

// lua_redispp.cpp
#include <sw/redis++/redis++.h>
#include <iostream>
using namespace sw::redis;
int main() {
    auto redis = Redis("tcp://127.0.0.1:6379");
    redis.set("stock", "100");
    std::string script = R"(
        local current = redis.call('GET', KEYS[1])
        if not current then return -1 end
        local stock = tonumber(current)
        local amount = tonumber(ARGV[1])
        if stock < amount then return -2 end
        redis.call('SET', KEYS[1], stock - amount)
        return stock - amount
    )";
    auto result = redis.eval<long long>(script, {"stock"}, {"5"});
    std::cout << "차감 후 재고: " << result << "\n";
    return 0;
}

EVALSHA로 스크립트 캐싱

동일 스크립트를 반복 실행할 때, SCRIPT LOAD로 SHA1 해시를 얻고 EVALSHA로 호출하면 네트워크 전송량이 줄어듭니다.

// hiredis EVALSHA
redisReply* r = (redisReply*)redisCommand(ctx, "SCRIPT LOAD %s", DECREMENT_STOCK);
std::string sha = r->str;
freeReplyObject(r);
r = (redisReply*)redisCommand(ctx, "EVALSHA %s 1 stock 5", sha.c_str());
freeReplyObject(r);

// redis-plus-plus: script_load로 SHA를 받고 evalsha로 호출
auto sha = redis.script_load(script);
auto left = redis.evalsha<long long>(sha, {"stock"}, {"5"});

Redis가 재시작되거나 다른 노드로 장애 조치되면 서버의 스크립트 캐시가 비어 EVALSHA가 NOSCRIPT 에러를 냅니다. 이때 EVAL로 다시 보내거나 SCRIPT LOAD를 다시 하는 폴백을 넣어 두어야 합니다.

분산 락 해제 Lua (같은 토큰일 때만 DEL)

-- unlock.lua
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else
    return 0
end

// C++에서 Lua 락 해제
const char* UNLOCK = "if redis.call('GET',KEYS[1])==ARGV[1] then return redis.call('DEL',KEYS[1]) else return 0 end";
redisReply* r = (redisReply*)redisCommand(ctx, "EVAL %s 1 lock:resource %s", UNLOCK, token.c_str());
freeReplyObject(r);

---

MULTI/EXEC 트랜잭션

트랜잭션이란?

MULTI로 트랜잭션을 시작하며, EXEC로 큐에 쌓인 명령을 한 번에 실행합니다. 중간에 DISCARD로 취소할 수 있습니다.

sequenceDiagram
    participant C as 클라이언트
    participant R as Redis
    C->>R: MULTI
    R->>C: OK
    C->>R: DECRBY stock 1
    R->>C: QUEUED
    C->>R: HSET order:123 item A
    R->>C: QUEUED
    C->>R: EXEC
    R->>C: [1, 1]  (순서대로 결과)

hiredis 트랜잭션

// transaction_hiredis.cpp
#include <hiredis/hiredis.h>
#include <iostream>
int main() {
    redisContext* ctx = redisConnect("127.0.0.1", 6379);
    if (!ctx || ctx->err) return 1;
    redisReply* r;
    r = (redisReply*)redisCommand(ctx, "MULTI");
    freeReplyObject(r);
    r = (redisReply*)redisCommand(ctx, "SET a 1");
    freeReplyObject(r);
    r = (redisReply*)redisCommand(ctx, "SET b 2");
    freeReplyObject(r);
    r = (redisReply*)redisCommand(ctx, "EXEC");
    if (r->type == REDIS_REPLY_ARRAY) {
        for (size_t i = 0; i < r->elements; ++i) {
            std::cout << "결과 " << i << ": " << r->element[i]->str << "\n";
        }
    }
    freeReplyObject(r);
    redisFree(ctx);
    return 0;
}

redis-plus-plus 트랜잭션

// transaction_redispp.cpp
#include <sw/redis++/redis++.h>
#include <iostream>
using namespace sw::redis;
int main() {
    auto redis = Redis("tcp://127.0.0.1:6379");
    auto tx = redis.transaction();
    tx.set("a", "1");
    tx.set("b", "2");
    auto replies = tx.exec();
    std::cout << "트랜잭션 완료: " << replies.size() << " 건\n";
    return 0;
}

주의: Redis 트랜잭션은 롤백을 지원하지 않습니다. EXEC 후 실행 단계에서 어떤 명령이 에러(예: 문자열 키에 HSET)를 내도 나머지 명령은 그대로 실행되고, 이미 실행된 명령은 되돌려지지 않습니다. Lua 스크립트도 마찬가지로, 스크립트 중간에 에러가 나면 그 전까지 실행한 쓰기는 남습니다. 둘 다 “다른 클라이언트가 중간에 끼어들지 않는다”는 의미의 원자성만 보장하므로, 조건 검사는 쓰기 전에 먼저 끝내고 검사를 통과한 경우에만 쓰도록 스크립트를 작성해야 합니다.

---

Redis Cluster와 redis-plus-plus

Cluster 개요

Redis Cluster는 데이터를 16384개 슬롯으로 나누어 여러 노드에 분산합니다. 키의 해시로 슬롯을 결정하며, 해당 슬롯을 가진 노드로 요청합니다.

flowchart TB
    subgraph Cluster[Redis Cluster]
        N1[노드 1: 0-5460]
        N2[노드 2: 5461-10922]
        N3[노드 3: 10923-16383]
    end
    C[클라이언트] --> N1
    C --> N2
    C --> N3

hiredis Cluster 한계

hiredis 단일 연결은 MOVED/ASK 리다이렉트를 자동 처리하지 않습니다. Cluster용으로는 hiredis 기반의 hiredis-cluster 라이브러리나 redis-plus-plus의 RedisCluster를 사용하는 것이 좋습니다.

redis-plus-plus RedisCluster

// cluster_redispp.cpp
#include <sw/redis++/redis++.h>
#include <iostream>
using namespace sw::redis;
int main() {
    // Cluster 노드 주소 (하나만 지정해도 됨)
    RedisCluster cluster("tcp://127.0.0.1:7000");
    cluster.set("key", "value");
    auto val = cluster.get("key");
    std::cout << "GET: " << *val << "\n";
    return 0;
}

주의: Cluster에서 다중 키 명령(MGET, MSET 등)은 키가 같은 슬롯에 있어야 합니다. {hash_tag}를 사용해 슬롯을 맞출 수 있습니다.

// 같은 슬롯: user:123:name, user:123:age
cluster.set("user:{123}:name", "홍길동");
cluster.set("user:{123}:age", "30");
cluster.mget({"user:{123}:name", "user:{123}:age"});

---

주문 알림·캐시 워밍업·재고 차감 예제

실시간 주문 알림 (Pub/Sub + 채널)

// order_notifier.cpp - 주문 완료 시 Pub/Sub으로 알림
#include <sw/redis++/redis++.h>
#include <atomic>
#include <iostream>
#include <thread>
#include <chrono>
#include <string>
using namespace sw::redis;
class OrderNotifier {
public:
    void start_subscriber() {
        sub_thread_ = std::thread([this]() {
            auto sub = redis_.subscriber();
            sub.on_message([](std::string channel, std::string msg) {
                std::cout << "[알림] " << channel << ": " << msg << "\n";
            });
            sub.subscribe("channel:orders");
            while (running_) {
                try {
                    sub.consume();
                } catch (const sw::redis::TimeoutError&) {
                    // socket_timeout마다 깨어나 running_을 확인
                }
            }
        });
    }
    void publish_order(const std::string& order_id) {
        redis_.publish("channel:orders", "주문 완료 #" + order_id);
    }
    ~OrderNotifier() {
        running_ = false;  // 구독 스레드가 다음 타임아웃에 루프를 빠져나옴
        if (sub_thread_.joinable()) sub_thread_.join();
    }
private:
    static ConnectionOptions make_opts() {
        ConnectionOptions opts;
        opts.host = "127.0.0.1";
        opts.port = 6379;
        opts.socket_timeout = std::chrono::milliseconds(200);  // consume()이 영원히 블로킹하지 않게
        return opts;
    }
    Redis redis_{make_opts()};
    std::atomic<bool> running_{true};
    std::thread sub_thread_;
};
int main() {
    OrderNotifier notifier;
    notifier.start_subscriber();
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    notifier.publish_order("12345");
    std::this_thread::sleep_for(std::chrono::seconds(2));
    return 0;
}

대량 캐시 워밍업 (파이프라인)

// cache_warmup.cpp - 파이프라인으로 10000건 SET
#include <sw/redis++/redis++.h>
#include <iostream>
#include <chrono>
#include <string>
using namespace sw::redis;
void warmup_cache(Redis& redis, int count) {
    auto pipe = redis.pipeline();
    for (int i = 0; i < count; ++i) {
        std::string key = "cache:product:" + std::to_string(i);
        std::string val = "{\"id\":" + std::to_string(i) + ",\"name\":\"상품" + std::to_string(i) + "\"}";
        pipe.set(key, val, std::chrono::seconds(300));
    }
    auto start = std::chrono::high_resolution_clock::now();
    pipe.exec();
    auto end = std::chrono::high_resolution_clock::now();
    auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
    std::cout << "파이프라인 " << count << "건 SET: " << ms << " ms\n";
}
int main() {
    auto redis = Redis("tcp://127.0.0.1:6379");
    warmup_cache(redis, 10000);
    return 0;
}

Lua로 원자적 재고 차감·분산 락

// atomic_stock.cpp - Lua로 재고 차감 + 락 해제
#include <sw/redis++/redis++.h>
#include <iostream>
#include <string>
using namespace sw::redis;
// Lua 테이블은 Redis 배열 응답으로 바뀌는데, 숫자는 정수 응답이 되므로
// C++에서 vector<string>으로 받으려면 tostring으로 모두 문자열로 맞춘다
const std::string DECREMENT_STOCK = R"(
local current = redis.call('GET', KEYS[1])
if not current then return {'-1', '키 없음'} end
local stock = tonumber(current)
local amount = tonumber(ARGV[1])
if stock < amount then return {'-2', '재고 부족'} end
redis.call('SET', KEYS[1], stock - amount)
return {tostring(stock - amount), 'OK'}
)";
int main() {
    auto redis = Redis("tcp://127.0.0.1:6379");
    redis.set("stock:item1", "100");
    auto result = redis.eval<std::vector<std::string>>(DECREMENT_STOCK, {"stock:item1"}, {"5"});
    std::cout << "차감 후: " << result[0] << ", " << result[1] << "\n";
    return 0;
}

---

SUBSCRIBE 연결 오용, CROSSSLOT, Lua nil: 에러 해결

SUBSCRIBE 연결에서 GET/SET 시도

증상: (error) ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context 원인: SUBSCRIBE한 연결은 구독 모드가 되어 일반 명령 사용 불가 해결법:

// ❌ 잘못된 사용 (hiredis): 구독한 연결로 일반 명령 전송
redisCommand(ctx, "SUBSCRIBE channel");
redisCommand(ctx, "GET key");  // 에러!
// ✅ 구독 전용 연결과 명령용 연결을 분리
auto redis = Redis("tcp://127.0.0.1:6379");   // 일반 명령·PUBLISH (연결 풀 사용)
auto sub = redis.subscriber();                 // 구독 전용 연결을 새로 만듦
sub.subscribe("channel");
redis.publish("channel", "msg");
redis.get("key");  // 문제없음

파이프라인 중간에 다른 명령

증상: 응답 순서가 꼬이거나 잘못된 결과 원인: redisAppendCommand와 redisGetReply 사이에 redisCommand 호출 해결법:

// ❌ 잘못된 사용
redisAppendCommand(ctx, "SET a 1");
redisCommand(ctx, "GET b");  // 파이프라인 깨짐
redisGetReply(ctx, &reply);
// ✅ 파이프라인은 한 번에
redisAppendCommand(ctx, "SET a 1");
redisAppendCommand(ctx, "SET b 2");
redisGetReply(ctx, &reply); freeReplyObject(reply);
redisGetReply(ctx, &reply); freeReplyObject(reply);

Lua 스크립트에서 nil 반환

증상: (error) ERR Error running script (call to f_xxx): @user_script:0: user_script:0: attempt to compare nil with number 원인: redis.call('GET', key)가 키가 없으면 nil 반환 해결법:

-- ❌ nil 체크 없음
local current = redis.call('GET', KEYS[1])
local stock = tonumber(current)  -- nil이면 에러
-- ✅ nil 체크
local current = redis.call('GET', KEYS[1])
if not current then
    return -1
end
local stock = tonumber(current)

Cluster에서 CROSSSLOT

증상: (error) CROSSSLOT Keys in request don't hash to the same slot 원인: MGET, MSET 등에서 키가 서로 다른 슬롯에 있음 해결법:

// ❌ 다른 슬롯
cluster.mget({"user:1:name", "user:2:name"});
// ✅ 같은 슬롯: {tag} 사용
cluster.mget({"user:{1}:name", "user:{1}:age"});

파이프라인 스레드 안전성

증상: redis-plus-plus에서 멀티스레드에서 Pipeline 사용 시 크래시 또는 잘못된 결과 원인: Pipeline은 스레드 안전하지 않음 해결법:

// ❌ 스레드 안전하지 않음
auto pipe = redis.pipeline();
std::thread t1([&]() { pipe.set("a", "1"); });
std::thread t2([&]() { pipe.set("b", "2"); });
// ✅ 스레드당 별도 Pipeline 또는 mutex
std::mutex mtx;
std::thread t1([&]() {
    std::lock_guard<std::mutex> lk(mtx);
    auto pipe = redis.pipeline();
    pipe.set("a", "1");
    pipe.exec();
});

사용자 입력을 스크립트 본문에 끼워 넣음

증상: 따옴표가 든 입력에서 Lua 문법 에러가 나거나, 의도하지 않은 Lua 코드가 실행됩니다. 원인: 사용자 입력을 문자열 연결로 스크립트 본문에 넣었습니다. 해결법: 입력은 항상 KEYS/ARGV로 넘깁니다. hiredis의 redisCommand에서 %s 자리에 넣은 값은 공백이나 따옴표가 있어도 길이가 붙은 별도 인자로 전송되므로 이스케이프가 필요 없습니다. 문제는 값을 스크립트 문자열이나 포맷 문자열 자체에 이어 붙일 때 생깁니다.

// ❌ 입력을 스크립트 본문에 연결
std::string bad = "return redis.call('GET', '" + user_input + "')";
// ✅ 스크립트는 고정, 값은 인자로
redisCommand(ctx, "EVAL %s 1 %s", "return redis.call('GET', KEYS[1])", user_input.c_str());
redis.eval<OptionalString>("return redis.call('GET', KEYS[1])", {user_input}, {});

---

배치 크기·짧은 Lua·Streams 대안

파이프라인 배치 크기

한 번에 너무 많은 명령을 쌓으면 클라이언트와 서버 양쪽에서 응답 버퍼가 커지고, 응답을 다 받을 때까지 결과를 쓸 수 없습니다. 수백~수천 건 단위로 나눠 보내면 왕복 절감 효과는 대부분 얻으면서 메모리 사용을 제한할 수 있으므로, 실제 값 크기로 배치 크기를 측정해 정합니다.

const int BATCH_SIZE = 200;
for (int offset = 0; offset < total; offset += BATCH_SIZE) {
    auto pipe = redis.pipeline();
    for (int i = 0; i < BATCH_SIZE && offset + i < total; ++i) {
        pipe.set(keys[offset + i], values[offset + i]);
    }
    pipe.exec();
}

Lua 스크립트는 짧고 단순하게

Lua 스크립트는 Redis를 블로킹합니다. 긴 스크립트는 다른 클라이언트에 지연을 줍니다. 복잡한 로직은 클라이언트에서 처리하며, 원자성이 필요한 부분만 Lua로 처리하세요.

Pub/Sub 대안 — Redis Streams

Pub/Sub은 메시지가 구독자에게만 전달되고 저장되지 않습니다. 메시지가 끊기면 유실됩니다. 영속성이 필요하면 Redis Streams를 사용하세요.

// Redis Streams: XADD로 기록, XREADGROUP으로 소비자 그룹 단위 처리
std::vector<std::pair<std::string, std::string>> fields = {
    {"event", "order_complete"}, {"id", "123"}};
auto id = redis.xadd("stream:orders", "*", fields.begin(), fields.end());

대량 조회 시 MGET + 파이프라인

MGET은 같은 슬롯(Cluster) 또는 단일 노드에서 여러 키를 한 번에 가져옵니다. 파이프라인과 함께 사용하면 대량 조회 시 효율적입니다.

// 1000개 키를 100개씩 MGET
for (size_t i = 0; i < keys.size(); i += 100) {
    auto end = std::min(i + 100, keys.size());
    std::vector<std::string> batch(keys.begin() + i, keys.begin() + end);
    auto vals = redis.mget(batch);
}

연결 풀 크기 조정

ConnectionOptions opts;
opts.host = "127.0.0.1";
opts.port = 6379;
ConnectionPoolOptions pool_opts;
pool_opts.size = 20;  // 동시 요청 수에 맞게 조정
pool_opts.wait_timeout = std::chrono::milliseconds(100);
auto redis = Redis(opts, pool_opts);

---

구독자 재연결·Lua 파일 관리·WATCH 낙관적 락

Pub/Sub + 구독자 재연결

구독 연결이 끊어지면 자동 재연결하는 루프를 두는 것이 좋습니다. Pub/Sub은 끊겨 있는 동안 발행된 메시지를 다시 받을 방법이 없으므로, 재연결 사이에 유실돼도 되는 알림에만 쓰고, 유실되면 안 되는 이벤트는 Streams를 씁니다.

void run_subscriber_with_reconnect(Redis& redis, const std::string& channel) {
    while (true) {
        try {
            auto sub = redis.subscriber();
            sub.on_message([](std::string ch, std::string msg) {
                std::cout << ch << ": " << msg << "\n";
            });
            sub.subscribe(channel);
            while (true) {
                sub.consume();  // 연결이 끊기면 예외가 나서 바깥 루프로 감
            }
        } catch (const std::exception& e) {
            std::cerr << "구독 재연결: " << e.what() << "\n";
            std::this_thread::sleep_for(std::chrono::seconds(5));
        }
    }
}

Lua 스크립트 파일로 관리

스크립트가 길어지면 .lua 파일로 분리하며, 빌드 시 포함하거나 런타임에 읽어옵니다.

std::string load_script(const std::string& path) {
    std::ifstream f(path);
    return std::string(std::istreambuf_iterator<char>(f), {});
}
auto script = load_script("scripts/decrement_stock.lua");
redis.eval<long long>(script, {"stock"}, {"5"});

트랜잭션 + WATCH (낙관적 락)

WATCH로 키를 감시하며, EXEC 시점에 값이 바뀌었으면 재시도합니다.

// hiredis WATCH
redisReply* r = (redisReply*)redisCommand(ctx, "WATCH stock");
freeReplyObject(r);
// WATCH 이후 GET으로 현재 값을 읽고 조건을 확인한 뒤
r = (redisReply*)redisCommand(ctx, "MULTI");
freeReplyObject(r);
// ... 명령 추가 (각 응답은 QUEUED) ...
r = (redisReply*)redisCommand(ctx, "EXEC");
if (r && r->type == REDIS_REPLY_NIL) {
    // WATCH한 키가 다른 클라이언트에 의해 바뀌어 트랜잭션이 취소됨 → 처음부터 재시도
}
freeReplyObject(r);

설정 외부화

struct RedisAdvancedConfig {
    std::string host = "127.0.0.1";
    int port = 6379;
    int pipeline_batch_size = 200;
    int pool_size = 10;
};
RedisAdvancedConfig load_config() {
    RedisAdvancedConfig c;
    if (const char* h = std::getenv("REDIS_HOST")) c.host = h;
    if (const char* p = std::getenv("REDIS_PORT")) c.port = std::stoi(p);
    return c;
}

---

자주 묻는 질문 (FAQ)

Q. Pub/Sub과 Streams 중 무엇을 써야 하나요?

A. Pub/Sub은 메시지를 저장하지 않고 그 순간 구독 중인 연결에만 전달하므로, 구독자가 없거나 연결이 끊겨 있으면 메시지는 사라집니다. 실시간 알림처럼 놓쳐도 다음 메시지로 회복되는 경우에 적합합니다. 처리 보장이 필요한 주문 이벤트 같은 데이터는 메시지를 저장하고 소비자 그룹별로 처리 위치(ACK)를 관리하는 Streams가 맞습니다.

참고 자료

• Redis 공식 문서 — Pub/Sub
• Redis 공식 문서 — Lua 스크립팅
• Redis 공식 문서 — 트랜잭션
• Redis 공식 문서 — Cluster
• redis-plus-plus GitHub
• hiredis GitHub

---

같이 보면 좋은 글

• C++에서 Redis 쓰기: hiredis·redis-plus-plus, 파이프라인, 분산 락, Lua 스크립트
• C++에서 DB 연동하기
• C++에서 PostgreSQL 연동

---