안녕하세요. 엔지엠소프트웨어입니다. 이전 시간에 카프카와 브로커 및 토픽에 대해 간단하게 알아봤는데요. 오늘은 프로듀서에 대해 알아보는 시간을 가지도록 하겠습니다. 우선, 이전 시간의 내용을 읽고 이 글을 보시는걸 추천 드립니다. 카프카에 대한 첫번째 글을 보려면, 아래 링크를 클릭하세요^^

[ 카프카란? (What is Kafka?) ]

 

프로듀서는 컨슈머에게 메세지를 보내는 역할을 합니다. 프로듀서가 보낸 메세지를 관리하는건 카프카 클러스터안에 브로커가 담당합니다. 아래는 C#으로 만든 간단한 프로듀서 예제입니다. 카프카 버전에 따라서 콘피그(Config)와 메소드(Method)가 다를 수 있습니다. 먼저 Config로 설정 정보를 만듭니다. 부트스트랩 서버를 설정하고, 메세지의 최대 크기를 설정 해줍니다. 메세지 전송에 필요한 프로듀서는 Headers와 Key, Value를 담아서 보낼 수 있습니다. 프로듀스(Produce) 메소드는 각각 처리할 수 있게 오버로딩되어 있으니 상황에 맞게 사용하면 됩니다.

try
{
    var config = new ProducerConfig
    {
        BootstrapServers = $"{model.Host}:{model.Port}",
        MessageMaxBytes = model.MessageMaxBytes,
        ApiVersionRequestTimeoutMs = model.ApiVersionRequestTimeoutMs
    };

    using (var producer = new ProducerBuilder<Null, string>(config).Build())
    {
        var message = new Message<Null, string> { Value = model.Message };
        producer.Produce(model.Topic, message);
    };
}
catch (Exception ex)
{
    if (model.MainView != null && model.MainView.ShowDebugOutput)
        player.WriteOutput($"Error　{ex.Message}");
}

 

메세지를 전송하면 최대 블록 사이즈를 확인한 후 시리얼라이즈된 데이타를 파티셔닝합니다. 프로듀서의 파티션이 결정되면, 버퍼의 배치에 메세지를 저장하고 메모리로 로딩된 메세지를 센더(Sender)가 카프카 브로커로 보내게됩니다. 이 때 어느 브로커의 파티션으로 보낼지 결정하게 됩니다. 아래 그림에서 Accumulator에서 배치에 모아진 데이타는 버퍼 메모리로 보내지는데요. 이 때 버퍼에 바로 저장하지 않고 배치를 묶어서 저장합니다.

 

 

Sender는 멀티 스레드로 동작하고, 배치에 메세지가 찼는지 여부에 상관없이 차례대로 브로커에 전송합니다. 다시 말해서 Produce 메소드는 계속해서 배치로 메세지를 보내고, Sender는 배치가 가득 차지 않아도 메세지를 보낼 수 있습니다. 이렇게 동작하는 이유는 linger.ms 옵션때문에 그렇습니다. 이 옵션은 전송 대기 시간을 말하는데요. 기본값은 0입니다. 대기 시간이 없으면 배치가 덜 차도 브로커로 바로 전송합니다. 만약, 대기 시간을 주면 그 시간만큼 기다렸다가 배치를 묶어서 전송합니다. 이 옵션으로 처리량을 늘리거나 줄일 수 있습니다. 예를 들어서 0.01초마다 들어오는 메세지가 있다고 가정 해보세요. 0.01초마다 메세지를 받아서 보내는 동작을 1초동안 100번 수행하는것보다 linger.ms 옵션을 10으로 주고, 1초동안 10번 수행하는게 속도면에서 더 좋습니다. 그만큼 불필요한 동작을 줄일 수 있기 때문입니다.

 

아래와 같이 토픽에 메세지를 보내면 전송 실패에 대한 여부를 알 수 없게됩니다. 이렇게 처리할 때는 메세지가 누락되거나 전송되지 않더라도 문제가 없는 시스템에 사용해야 합니다.

producer.Produce(model.Topic, message);

 

프로듀서 빌더에 에러 핸들러를 추가해주면, 메세지를 전송하지 못한 경우 어떤 문제인지 확인할 수 있습니다. 이전에는 동기적으로 처리가 되었는데요. 지금은 Callback 함수로 처리할 수 있게 되었습니다. 아마도, 메세지를 동기적으로 처리하려다보니 성능 저하가 문제가 되었을거 같아요. 비동기로 처리하면 대기가 발생하지 않기 때문에 효율적으로 처리할 수 있게됩니다.

try
{
    var config = new ProducerConfig
    {
        BootstrapServers = $"{model.Host}:{model.Port}",
        MessageMaxBytes = model.MessageMaxBytes,
        ApiVersionRequestTimeoutMs = model.ApiVersionRequestTimeoutMs
    };

    using (var producer = new ProducerBuilder<Null, string>(config).SetErrorHandler((s, e) =>
            {
                if (model.MainView != null && model.MainView.ShowDebugOutput)
                    player.WriteOutput($"Error　{e.Reason}");
            }).Build())
    {
        var message = new Message<Null, string> { Value = model.Message };
        producer.Produce(model.Topic, message);
    };
}
catch (Exception ex)
{
    if (model.MainView != null && model.MainView.ShowDebugOutput)
        player.WriteOutput($"Error　{ex.Message}");
}

 

프로듀서는 전송을 보장하기 위해 Ack값을 사용합니다. Ack가 None이면 서버 응답을 기다리지 않기 때문에 메세지 전송 여부를 확인할 수 있는 방법이 없습니다. Ack를 Leader로 설정하면 파티션의 리더에 저장되면 응답을 받습니다. 리더에 장애가 발생하면 메세지가 유실될 가능성이 있습니다. Ack를 All로 설정하면 모든 리플리카에 저장되면 응답을 받습니다. 이 설정은 브로커의 min.insync.replicas 설정에 따라 처리가 달라집니다.

 

 

Ack와 min.insync.replicas 설정은 프로듀서 Ack 옵션이 All일 때 저장에 성공했다고 응답할 수 있는 최소 리플리카의 갯수입니다. 예를 들어서 토픽을 만들 때 리플리카 갯수를 3으로 설정하고 Ack는 All, min.insync.replicas를 2로 설정하면 리더에 저장하고, 팔로워중 한개에 저장하면 메세지 전송이 성공했다고 응답을 줍니다. 만약, 리플리카가 3이고, Ack가 All인 상태에서 min.insync.replicas를 1로 설정하면 어떻게 될까요? 리더에 저장되면 성공 응답을 보내주지만 리더에 장애가 발생하면 메세지를 유실할 가능성이 있습니다. 그리고, 동일한 설정에서 min.insync.replicas를 3으로 설정하면 리더와 팔로워 2개에 저장되면 성공합니다. 그런데 팔로워중 한개라도 장애가 발생하면 리플리카 부족으로 메세지 저장에 실패하게 됩니다. 그래서, 이런 설정을 적절하게 잘 사용해야 문제없이 쓸 수 있습니다.

 

아무래도 전송을 실패하는 여러가지 요인이 있겠지만, 전송 과정에서 문제가되는 시나리오는 보통 아래와 같습니다.

• 전송 타임 아웃 (네트워크 문제거나 버그)
• 리더 다운에 의해 새로운 리더가 선출되는 중
• 브로커 설정 메세지 크기 한도 초과
• 메모리 부족등등...

 

프로듀서가 전송을 실패한다면 위와같이 Error Handler를 설정하고, 에러 메세지 내용에 따라서 적절하게 처리해줘야 합니다. 대부분의 메세지 버스 솔루션이 동일한 기능을 제공해주는데요. 설정에 따라 다르겠지만, 자체적으로 메세지를 다시 재시도 해주게 됩니다. 여기까지는 문제가 될거 같지 않은데요. 만약, try ~ catch에서 무한으로 메세지를 재전송하게 하면 시스템이 다운될 수 있습니다. 보낼 메세지는 계속해서 들어오는데 문제가 되는 메세지를 보내기 위해 계속 리트라이하기 때문입니다. 이런 경우 몇번의 리트라이 후 관리자에게 에러 메세지를 보내거나 알람을 발생시켜야 합니다. 일부 유실에 대해서 문제가 되지 않는다면 3회 또는 5회 시도 후 건너뛰도록 하세요^^

 

더 많은 옵션들과 설정들이 존재하는데요. 기본적으로 사용되는 내용들만 간단하게 알아봤습니다. 회사 또는 프로젝트마다 환경이 다르기도 하고, 장애가 발생할 수 있는 시나리오도 다를거거든요. 이런 것들은 프로젝트를 진행하면서 예측하지 못한 문제점들을 확인하면서 설정을 수정해야 할수도 있습니다. 다음에는 좀 더 재미있는 내용을 소개할 수 있으면 좋겠네요.

 

개발자에게 후원하기

 

추천, 구독, 홍보 꼭~ 부탁드립니다.

여러분의 후원이 빠른 귀농을 가능하게 해줍니다~ 답답한 도시를 벗어나 귀농하고 싶은 개발자~

감사합니다~

추천0 비추천0