메시징

메시징 (Messaging): 메시지·토픽·구독

Pulsar는 게시-구독(publish-subscribe, 줄여서 pub-sub) 패턴 위에 설계된 시스템이에요. 이 패턴에서 프로듀서(producer)는 토픽으로 메시지를 발행하고, 컨슈머(consumer)는 그 토픽을 구독해 메시지를 받아 처리한 뒤 브로커에 확인(acknowledgment)을 보내요. 구독이 만들어지면 컨슈머가 끊겨 있어도 Pulsar는 메시지를 모두 보관하고, 컨슈머가 성공적으로 처리했다고 확인한 메시지만 삭제해요.

출처: Messaging - Apache Pulsar 공식 문서

본문

메시지 (Messages)

메시지는 Pulsar의 기본 '단위'예요. 프로듀서가 토픽에 발행하고 컨슈머가 토픽에서 소비하는 대상이 바로 메시지죠. 메시지가 가진 구성 요소를 표로 정리할게요.

메시지 기본 최대 크기는 5MB예요. 이 값은 broker.conf의 maxMessageSize(기본 5242880)와 bookkeeper.conf의 nettyMaxFrameSizeBytes(기본 5253120)로 조정할 수 있어요.

확인 (Acknowledgment)

컨슈머가 메시지를 성공적으로 소비한 뒤 브로커에 확인을 보내면, 그 메시지는 모든 구독이 확인하기 전까지 영구 저장되고, 확인된 뒤에야 삭제돼요. 확인(ack)은 곧 "이 메시지는 시스템에서 지워도 된다"는 신호예요. 이미 확인된 메시지까지 저장하고 싶다면 메시지 보존 정책(message retention policy)을 설정해야 해요.

메시지를 확인하는 방식은 두 가지예요.

• 개별 확인(individual acknowledgment): 컨슈머가 메시지 하나씩 확인 요청을 보내요. consumer.acknowledge(msg) API를 써요.
• 누적 확인(cumulative acknowledgment): 컨슈머가 마지막으로 받은 메시지만 확인해요. 그 메시지까지 포함해서 스트림의 모든 메시지는 그 컨슈머에게 재전송되지 않아요. consumer.acknowledgeCumulative(msg)를 써요.

누적 확인은 Shared 또는 Key_Shared 구독 타입에서는 쓸 수 없어요. 여러 컨슈머가 같은 구독에 접근하기 때문에, 그 타입에서는 반드시 개별 확인을 해야 해요.

부정 확인 (Negative acknowledgment)

부정 확인(nack)은 컨슈머가 메시지를 처리하지 못했음을 브로커에 알리는 메커니즘이에요. 컨슈머가 메시지를 소비하다 실패해서 다시 소비해야 할 때 부정 확인을 보내면 브로커가 그 메시지를 다시 전달해요. Exclusive·Failover 타입에서는 마지막으로 받은 메시지에 대해서만, Shared·Key_Shared 타입에서는 메시지 개별로 부정 확인할 수 있어요.

Consumer<byte[]> consumer = pulsarClient.newConsumer()
    .topic(topic)
    .subscriptionName("sub-negative-ack")
    .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
    .negativeAckRedeliveryDelay(2, TimeUnit.SECONDS) // 기본값은 1분
    .subscribe();
Message<byte[]> message = consumer.receive();
consumer.negativeAcknowledge(message); // 부정 확인 API 호출

주의할 점이 하나 있어요. negativeAcknowledge의 재전송 카운터는 메모리에만 보관돼서 브로커가 구독 디스패처를 닫으면(재시작, bundle unload, topic unload, 컨슈머 연결 해제) 0으로 초기화돼요. 그래서 DeadLetterPolicy.maxRedeliverCount에 도달하지 못하고 실패 메시지가 무한 재전송될 수 있어요. 신뢰성 있는 재시도 제한이 필요하다면 재시도 횟수를 메시지 속성으로 영구 보존하는 reconsumeLater + enableRetry(true)를 쓰는 게 좋아요.

확인 타임아웃 (Acknowledgment timeout)

기본적으로 확인 타임아웃은 비활성화라서, 컨슈머가 죽지 않는 한 전달된 메시지는 재전달되지 않아요. 확인 타임아웃(ackTimeout)을 설정하면 클라이언트가 미확인 메시지를 추적하다가, 그 시간이 지나면 브로커에 미확인 메시지 재전달 요청을 보내요. 일반적으로는 부정 확인이 확인 타임아웃보다 선호돼요. 타임아웃 값을 잡기 어렵기도 하고, 타임아웃이 지났다고 해서 그 메시지가 꼭 다시 소비돼야 하는 건 아니기 때문이에요.

consumer.ackTimeout(2, TimeUnit.SECONDS) // 기본값은 0
    .ackTimeoutTickTime(1, TimeUnit.SECONDS)
    .subscriptionName("sub")
    .subscribe();

재시도 편지 토픽 (Retry letter topic)

재시도 편지 토픽은 소비에 실패한 메시지를 저장해 두고 나중에 다시 소비하게 해줘요. 이 방식으로 메시지가 재전달되는 간격을 커스터마이징할 수 있어요. 원래 토픽의 컨슈머는 재시도 편지 토픽에도 자동으로 구독돼요. 최대 재시도 횟수에 도달하면 소비되지 못한 메시지는 수동 처리를 위해 데드 레터 토픽으로 옮겨져요.

기본적으로 자동 재시도는 비활성화돼 있고, enableRetry(true)로 켤 수 있어요. 기본 재시도 편지 토픽 이름 형식은 <topicname>-<subscriptionname>-RETRY예요. 재시도 편지 토픽의 메시지는 클라이언트가 자동으로 만드는 특별 속성을 가지는데, REAL_TOPIC, ORIGIN_MESSAGE_ID, RECONSUMETIMES, DELAY_TIME이 대표적이에요.

consumer.reconsumeLater(msg, 3, TimeUnit.SECONDS);

재시도 편지 토픽은 reconsumeLater + enableRetry(true)로 쓸 때 재시도 횟수가 메시지 속성으로 영구 보존돼요. 그래서 maxRedeliverCount를 확실히 지키고 데드 레터 토픽까지 도달하게 만들 수 있는 유일한 메커니즘이에요.

데드 레터 토픽 (Dead letter topic)

데드 레터 토픽은 일부 메시지를 성공적으로 소비하지 못해도 소비를 계속할 수 있게 해줘요. 소비에 실패한 메시지를 특정 토픽(데드 레터 토픽)에 저장해 두고, 그 메시지를 어떻게 처리할지는 사용자가 정해요. 기본 데드 레터 토픽 이름 형식은 <topicname>-<subscriptionname>-DLQ예요. 데드 레터 토픽은 Shared와 Key_Shared 구독 타입에서 활성화할 수 있어요.

메시지 압축 (Compression)

메시지 압축은 일부 CPU 오버헤드를 지불하는 대신 메시지 크기를 줄여줘요. Pulsar 클라이언트는 LZ4, ZLIB, ZSTD, SNAPPY 압축 타입을 지원해요. 압축 타입은 메시지 메타데이터에 저장되므로 컨슈머는 필요에 따라 다른 압축 타입을 자동으로 적용할 수 있어요.

client.newProducer().topic("topic-name").compressionType(CompressionType.LZ4).create();

배치 (Batching)

배치가 활성화되면 프로듀서는 메시지 여러 개를 모아 하나의 요청으로 보내요. 배치 크기는 최대 메시지 수와 최대 publish latency로 결정돼요. 따라서 backlog 크기는 메시지 수가 아니라 배치 수를 나타내요. Pulsar에서 배치는 개별 메시지가 아니라 단일 단위로 추적·저장되고, 컨슈머가 이를 개별 메시지로 풀어내요. 일반적으로 배치는 그 안의 메시지가 전부 확인될 때 확인되어요. 그래서 배치 전체가 확인되지 않으면 예상치 못한 실패·부정 확인·확인 타임아웃으로 배치의 모든 메시지가 재전달될 수 있어요.

토픽 (Topics)

Pulsar에서는 토픽을 명시적으로 만들 필요가 없어요. 클라이언트가 아직 존재하지 않는 토픽에 메시지를 쓰거나 받으려 하면, Pulsar가 토픽 이름에 담긴 네임스페이스 아래에 그 토픽을 자동으로 생성해요.

네임스페이스 (Namespaces)

네임스페이스는 애플리케이션이 토픽의 계층을 만들고 관리하게 해줘요. 예를 들어 my-tenant/app1이라는 토픽은 my-tenant의 app1 애플리케이션을 위한 네임스페이스예요. 네임스페이스 아래에는 원하는 만큼 토픽을 만들 수 있어요.

구독 (Subscriptions)과 구독 타입

구독은 컨슈머가 메시지를 받는 방식을 결정해요. 대표적인 타입 두 가지를 살펴볼게요.

• Failover: 비파티션 토픽이 여러 개일 때 컨슈머는 컨슈머 이름 해시와 토픽 이름 해시로 선택돼요. 클라이언트는 같은 컨슈머 이름으로 모든 토픽을 구독해요.
• Key_Shared: 메시지 키(또는 ordering key)를 해시 함수(예: Murmur3 32-bit)에 넣어 32비트 정수 해시를 만들고, 그 해시로 연결된 컨슈머 중 하나를 골라요. 같은 키를 가진 메시지는 한 번에 한 컨슈머에게만 전달되고 미확인 상태로 남을 수 있어서, 키별 전달 순서가 보존돼요.

Key_Shared 구독에서는 키 기반 배치(Key_Shared batching)를 쓸 수 있어요. BatcherBuilder.KEY_BASED로 프로듀서를 만들면 돼요.

Producer<byte[]> producer = client.newProducer()
    .topic("my-topic")
    .batcherBuilder(BatcherBuilder.KEY_BASED)
    .create();

멀티 토픽 구독 (Multi-topic subscriptions)

여러 토픽을 구독할 때 Pulsar 클라이언트는 regex 패턴이나 목록과 일치하는 토픽을 찾기 위해 자동으로 Pulsar API를 호출하고, 모두 구독해요. 토픽 중 일부가 아직 없더라도 생성되는 즉시 자동으로 구독해요. 예를 들어 Pattern.compile("persistent://.*/my-topic-.*") 같은 정규식으로 네임스페이스 안의 토픽들을 한 번에 구독할 수 있어요.

파티션 토픽과 해싱 (Partitioned topics & Hashing)

파티션 토픽에서는 특정 메시지를 어느 파티션으로 보낼지 해싱 함수가 정해요. 표준 해싱 함수는 두 가지예요.

• JavaStringHash
• Murmur3_32Hash

프로듀서의 기본 해싱 함수는 JavaStringHash예요. 다만 서로 다른 여러 언어 클라이언트가 함께 쓰이는 환경에서는 JavaStringHash가 유용하지 않으니, 그런 경우 Murmur3_32Hash를 권장해요.

비영구 토픽 (Non-persistent topics)

프로듀서와 컨슈머는 영구 토픽과 같은 방식으로 비영구 토픽에 연결할 수 있어요. 결정적인 차이는 토픽 이름이 non-persistent로 시작해야 한다는 점이에요. exclusive, shared, key_shared, failover 모든 구독 타입이 비영구 토픽에서 지원돼요.

지연 메시지 전달 (Delayed message delivery)

메시지를 지정된 시간 뒤에 전달하고 싶을 때 쓰는 기능이에요. 재시도 편지 토픽이 "실패한 메시지를 재전달해 데이터 손실을 막는 것"을 목적으로 한다면, 지연 메시지 전달은 "메시지를 지정된 딜레이 후에 전달하는 것"을 목적으로 해요. 둘 다 나중에 소비한다는 점은 같지만 쓰임새가 달라요.

더 알아보기

• 프로듀서와 컨슈머, 리더 클라이언트가 실제로 어떻게 동작하는지는 Pulsar Clients 문서를 보면 돼요.
• Pulsar가 어떤 구조로 돌아가는지 살펴보려면 개요 문서부터 시작하는 걸 추천해요.
• 토픽 데이터의 형태를 합의하는 방법은 스키마 이해하기 문서에서 설명해요.