[C언어] Redis를 활용한 Pub/Sub 기능

앞서 구성했던 채팅 프로그램에서 채팅을 Redis의 Pub/Sub 기능을 이용하여 주고받을 수 있도록 구성하였다.

 

[K8S] C 기반 HTTP 통신 구현 및 Kubernetes 배포

 

 

Redis는 데이터를 Key-Value 형태로 메모리에 저장하고, 애플리케이션이 TCP 연결을 통해 명령을 보내는 서버이다.

C 서버
  │
  │ TCP 연결: redis-service:6379
  ▼
Redis
  │
  │ SET, GET, HSET, SADD, PUBLISH 등의 명령 처리
  ▼
C 서버에 처리 결과 반환

 

 

Redis는 String 자료형 말고도 Hash, Set, List, Stream 같은 자료형도 지원한다. Zamtok 에서는 다음과 같이 Reids 자료형을 구분하여 사용하고 있다.

자료형	저장 데이터	Redis Key
String	MDN과 사용자 ID의 매핑	mdn:<mdn>
Hash	사용자 및 채팅방 정보	user:<user_id>, room:<room_id>
Set	친구, 채팅방 참여 관계	friends:<user_id>, room_members:<room_id>
Pub/Sub	실시간 알림 전달	user:<user_id> 채널

 

 

C에서 Redis를 사용하려면 일반적으로 hiredis 라이브러리를 사용한다.

hiredis guide (C)

 

 

hiredis 설치

 

 

Zamtok은 쿠버네티스 환경에서 동작하기 때문에 별도 Redis를 설치할 필요없이 사용 가능하다. 단, Redis Service가 띄워져 있어야 한다.

 

Redis 연결

 

 

redisConnect()를 호출하면 애플리케이션과 Redis 서버 사이에 TCP 연결이 생성되며, redisContext *로 표현된다.

typedef struct redisContext {
    int err;            /* 에러 코드 (0이면 정상) */
    char errstr[128];   /* 에러 메시지 */
    int fd;             /* 실제 소켓 fd */
    ...
} redisContext;

 

redisContext는 Redis 연결 정보를 관리하는 hiredis 구조체로, 다음과 같은 정보가 들어 있다.

• Redis와 연결된 TCP 소켓
• Redis 서버 주소와 포트
• 송수신 버퍼
• 연결 오류 상태
• 오류 메시지

redisContext *redisConnect(const char *ip, int port);

 

redisConnect를 호출하면 내부적으로 socket() → connect()를 호출해 TCP 3-way handshake를 하고 redisContext*를 malloc해서 반환한다. 실패 케이스는 다음 두가지이며, ctx→errstr에 에러 메시지가 담겨있어 확인이 가능하다.

• NULL 반환 : malloc이 실패한 경우
• ctx→err != 0 : connect()가 실패한 경우 (호스트 없음, 포트 거부 등)

반환값인 redisContexts*는 hiredis가 malloc으로 만든 것이기 때문에 Redis 연결을 종료할 때는 redisFree(ctx)를 호출해주어야 한다.

 

 

다음은 Redis 서버로 명령을 전송하고 처리 결과를 받아오는 함수이다.

void *redisCommand(redisContext *context, const char *format, ...);

• *context : 어떤 Redis 연결을 사용할지 지정한다. 
• *format : Redis에 전송할 명령을 작성한다.

redisCommand()는 내부적으로 다음 작업을 수행한다.

1. 전달받은 문자열 (*format) 을 Redis 통신 프로토콜로 변환한다.
2. redisCotext가 관리하는 TCP 소켓으로 명령을 전송한다.
3. Redis 서버의 응답을 기다린다.
4. 응답을 redisReply 구조체로 만들어 반환한다.

 

Redis Reply

 

redisReply는 Redis 서버 응답을 담는 구조체이다.

<hiredis/hiredis.h>

typedef struct redisReply {
    int type;                    /* REDIS_REPLY_* 중 하나 */
    long long integer;           /* type == REDIS_REPLY_INTEGER 일 때 값 */
    size_t len;                  /* str의 길이 */
    char *str;                   /* type == REDIS_REPLY_STRING/ERROR/STATUS 일 때 문자열 */
    size_t elements;             /* type == REDIS_REPLY_ARRAY 일 때 하위 요소 개수 */
    struct redisReply **element; /* type == REDIS_REPLY_ARRAY 일 때 하위 redisReply* 배열 */
} redisReply;

 

Redis 명령마다 응답 형태가 다르기 때문에 type부터 확인해야 한다.

Type	Desc	결과 위치
REDIS_REPLY_STATUS	상태 문자열 (OK 등)	reply->str
REDIS_REPLY_INTEGER	정수	reply->str
REDIS_REPLY_STRING	문자열	reply->integer
REDIS_REPLY_ARRAY	배열, 다중 응답	reply->element[]
REDIS_REPLY_NIL	키/필드 없음	없음
REDIS_REPLY_ERROR	서버 에러	reply->str

 

redisCommand가 리턴한 redisReply*은 hiredis가 malloc으로 만든 것이기 때문에 반드시 freeReplyObject()로 해제를 해야 한다. REDIS_REPLY_ARRAY인 경우 최상위 reply만 한 번 freeReplyObject를 해주면 된다.

 

---

 

 

Redis Pub/Sub 에서 채널은 메시지를 전달하기 위한 논리적인 통로이다. 발행자(Publisher)는 특정 채널로 메시지를 보내면, 해당 채널을 구독 중인 모든 구독자(Subscriber)는 메시지를 전달받는다.

 

여기서 주의해야 할 점은 SUBSCRIBE는 커넥션을 지속적으로 메시지를 수신하는 구독 상태로 전환한다.

 

즉, 한 커넥션에서 SUBSCRIBE를 실행하면 그 커넥션은 이후 pub/sub 명령 외에는 사용하지 못한다. 그렇기 때문에 일반 명령용과 구독 전용 두 개로 Redis 연결을 분리해야 한다.

int main( int argc, char **argv )
{
    const char *redis_host = getenv("REDIS_HOST");
    const char *redis_port_str = getenv("REDIS_PORT");
    int redis_port = redis_port_str ? atoi(redis_port_str) : 6379;

    if (!redis_host)
    	redis_host = "127.0.0.1";
        
    LOG_MSG("[INFO] Redis: %s:%d\n", redis_host, redis_port);
    
    rc = redis_connect(redis_host, redis_port);
    if ( rc != 0 )
    	LOG_MSG("[WARN] Redis connect fail (join/login will fail)\n");
    else
    	LOG_MSG("[INFO] Redis connected\n");

    rc = redis_pubsub_connect(redis_host, redis_port);
    if ( rc != 0 )
    	LOG_MSG("[WARN] Redis pubsub connect fail (user sub/pub will fail)\n");
    else
    	LOG_MSG("[INFO] Redis pubsub connected\n");
}

 

 

Subscribe

 

Zamtok에서 각 서버 파드는 자신에게 연결된 사용자가 메시지를 받을 수 있도록 Redis 채널을 구독한다. 채널 구독은 다음 두 가지 경우로 이루어진다.

 

SUBSCRIBE user:<user_id>

 

사용자가 처음 접속하여 특정 서버 파드와 연결되면, 해당 파드는 사용자의 <user_id>에 대응하는 개인 채널을 구독한다. user 채널은 특정 사용자에게만 전달해야 하는 알림이나 메시지를 처리할 때 사용한다.

예시 코드는 다음과 같다.

int redis_user_sub(const char *pstr_user_id, int i_len)
{
    redisReply *pt_reply = NULL;
    char str_user_channel[CHANNEL_MAX_LEN] = "";

    if (g_pubsub_ctx == NULL || pstr_user_id == NULL)
        return ZT_RC_ARG_INVALID;

    if (i_len <= 0 || i_len > USER_ID_MAX_LEN)
        return ZT_RC_ARG_INVALID;

    /* user channel */
    snprintf(str_user_channel, sizeof(str_user_channel), "user:%s", pstr_user_id);

    pt_reply = redisCommand(g_pubsub_ctx, REDIS_SUBSCRIBE_CMD, str_user_channel);
    if (pt_reply == NULL || pt_reply->type == REDIS_REPLY_ERROR) 
    {
        if (pt_reply)
            freeReplyObject(pt_reply);
        LOG_MSG("[redis] sub user fail: %s\n", str_user_channel);
        return ZT_RC_REDIS;
    }

    if (pt_reply->type != REDIS_REPLY_ARRAY || pt_reply->elements < 3 ||
        pt_reply->element[0]->str == NULL ||
        strcmp(pt_reply->element[0]->str, "subscribe") != 0) {
        freeReplyObject(pt_reply);
        LOG_MSG("[redis] sub user fail: %s\n", str_user_channel);
        return ZT_RC_REDIS;
    }

    freeReplyObject(pt_reply);
    LOG_MSG("[redis] sub user: %s\n", str_user_channel);

    return ZT_RC_OK;
}

 

 

 

SUBSCRIBE room:<room_id>

 

사용자가 새 채팅방을 만들거나 기존 채팅방에 입장하면, 사용자와 연결된 서버 파드는 해당 채팅방의 채널을 구독한다. room 채널은 채팅방에 참여한 사용자들에게 메시지를 전달할 때 사용한다.

 

예를 들어, 1001번 유저가 파드 A에 로그인해 있고, 1002번 유저는 파드 B에 로그인해 있다고 가정하자.

 

1. 클라이언트 (1001) → Pod A

 

1001번 유저가 방 room_1001에 1002번 유저를 초대한다.

POST /room/invite HTTP/1.1
Host: localhost:8080
Content-Type: application/json
Content-Length: 61

{"room_id": "room1_1001", "user_id": "1001", "members": "1002"}

 

2. Pod A → 클라이언트 (1001)

HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 60

{"success":true,"code":"SUCCESS","message":"ok","data":{"room_id":"room1_1001"}}

 

3.  Pod A → Redis

PUBLISH user:1002 {"type":"push","event":"room_invite","data":{"room_id":"room1_1001","from":"1001"}}

 

4. Redis → Pod B

["message", "user:1002", "{\"type\":\"push\",\"event\":\"room_invite\",\"data\":{\"room_id\":\"room1_1001\",\"from\":\"1001\"}}"]

 

5. Pod B → 클라이언트 (1002)

{"type":"push","event":"room_invite","data":{"room_id":"room1_1001","from":"1001"}}

 

 

UNSUBSCRIBE room:<room_id>
UNSUBSCRIBE user:<user_id>

 

사용자가 채팅방에서 나가거나 서버 연결을 종료하면 파드는 더 이상 필요하지 않은 채널의 구독을 해제한다. 단, 같은 파드에 room:<room_id>를 사용하는 다른 사용자가 남아 있다면 해당 채널의 구독을 유지해야 한다.

 

 

Publish

 

 

Zamtok에서 각 서버 파드는 자신에게 연결된 사용자가 메시지를 받을 수 있도록 Redis 채널을 구독한다. 채널 구독은 다음 두 가지 경우로 이루어진다.

 

PUBLISH를 실행하면 Redis는 메시지를 전달한 후 숫자 응답을 반환한다. 만약 다음과 같이 반환을 한다면 한 개의 구독 연결에 메시지가 전달됐다는 의미이다.

(integer) 1

 

 

 

다음 함수는 PUBLISH 된 메시지를 실시간으로 받기 위한 상시 수신 루프이다. 스레드의 흐름은 다음과 같다.

 

1. redisGetReply()에서 대기한다.
2. 다른 파드에서 PUBLISH 명령을 실행한다.
3. Redis가 채팅방 채널로 메시지를 전달한다.
4. redisGetReply()가 메시지를 redisReply 형태로 반환한다.
5. 채널과 payload를 redis_pubsub_dispatch()에 전달한다.
6. 응답 객체를 해제한 후 다음 메시지를 다시 기다린다.

static void *redis_pubsub_listen_thread(void *arg)
{
    redisReply *pt_reply = NULL;

    while (g_pubsub_ctx != NULL) {
        if (redisGetReply(g_pubsub_ctx, (void **)&pt_reply) != REDIS_OK) {
            LOG_MSG("[redis] pubsub listen: getReply fail\n");
            break;
        }

        if (pt_reply == NULL)
            continue;

        /* pub/sub 응답 형식: ["message", channel, payload] */
        if (pt_reply->type == REDIS_REPLY_ARRAY && pt_reply->elements >= 3 &&
            pt_reply->element[0]->str != NULL &&
            strcmp(pt_reply->element[0]->str, "message") == 0) {
            redis_pubsub_dispatch(pt_reply->element[1]->str, pt_reply->element[2]->str);
        }

        freeReplyObject(pt_reply);
        pt_reply = NULL;
    }

    return NULL;
}

 

 

redisGetReply

 

 

redisGetReply()는 hiredis의 동기식 Redis 연결에서 서버가 보내는 다음 응답 하나를 읽어 redisReply 객체로 반환하는 함수이다.

int redisGetReply(redisContext *context, void **reply);

 

 

redisGetReply()는 Redis 서버로부터 응답이 올 때까지 대기하는 블로킹 함수이다. 즉, Main 스레드에서 실행하면 메시지가 없는 동안에는 다른 처리가 멈추기 때문에 이를 방지하기 위해서 구독 전용 스레드가 필요하다.

 

ZamTok에서는 구독 전용 스레드를 생성하고, 해당 스레드에서 redisGetReply()를 반복적으로 호출한다

 

일반적으로 redisGetReply()가 받는 응답 형식은 Redis 명령에 따라 문자열, 숫자, 배열 등으로 달라진다. 하지만 SUBSCRIBE 상태에서 일반 채널 메시지를 받으면 다음과 같은 배열 형태의 응답이 전달된다.

배열	값	의미
element[0]	"message"	응답 종류
element[1]	"room:100"	채널
element[2]	"hello"	메시지

 

받은 메시지에서 채널 이름을 기준으로 user 채널인지 room 채널인지 구분할 수 있다.

메시지 처리가 끝난 후에는 redisGetReply()가 생성한 응답 객체를 반드시 해제해야 한다.