프로듀서 사용하기

프로듀서 사용하기 (Producers)

클라이언트를 설정한 뒤에는 프로듀서를 만들어 메시지를 게시할 수 있어요. 이 문서는 프로듀서를 만들고 이름을 지정하며, 메시지를 동기·비동기로 게시하고 다양한 메시지 속성을 설정하는 방법을 정리했어요. 파티셔닝된 토픽에서의 라우팅, 청킹, 인터셉터, 접근 모드 같은 고급 기능도 함께 살펴볼게요.

출처: 문서

본문

클라이언트를 설정한 뒤에는 더 살펴보며 프로듀서로 작업을 시작할 수 있어요.

프로듀서 만들기 (Create the producer)

이 예시는 프로듀서를 만드는 방법을 보여줘요.

• Java C++ Python

Java로 프로듀서를 동기적으로 만들기: Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic("my-topic").create(); Java로 프로듀서를 비동기적으로 만들기: pulsarClient.newProducer(Schema.STRING).topic("my-topic").createAsync().thenAccept(p -> { log.info("Producer created: {}", p.getProducerName()); });

Producer producer;
Result result = client.createProducer("my-topic", producer);

producer = client.create_producer('my-topic')

프로듀서 이름 지정 (Producer naming)

모든 프로듀서에는 모든 Pulsar 클러스터에서 고유해야 하는 이름이 있어요. 이름을 명시적으로 설정하지 않으면 Pulsar가 전역적으로 고유한 이름을 자동으로 생성해요. 이름을 지정하면 브로커가 그 이름을 가진 프로듀서 하나만 한 번에 토픽에 게시할 수 있도록 강제해요.

메시지 중복 제거(deduplication)를 사용할 때는 명시적인 프로듀서 이름을 설정해야 해요. 중복 제거가 필요하지 않더라도 의미 있는 프로듀서 이름을 설정하는 것이 권장돼요 — 브로커 로그, 관리자 통계, 메트릭에 이름이 나타나므로 디버깅이 훨씬 쉬워지고, 메시지를 만든 애플리케이션까지 빠르게 추적할 수 있거든요.

• Java C++ Python

Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
              .topic("my-topic")
              .producerName("my-unique-producer-name")
              .create();

ProducerConfiguration producerConfig;
producerConfig.setProducerName("my-unique-producer-name");
Producer producer;
Result result = client.createProducer("my-topic", producerConfig, producer);

producer = client.create_producer('my-topic', producer_name='my-unique-producer-name')

메시지 게시 (Publish messages)

Pulsar는 대부분의 클라이언트에서 메시지의 동기 및 비동기 게시를 모두 지원해요. Node.js, C# 같은 일부 언어별 클라이언트에서는 언어별 메커니즘(예: await)을 사용해 비동기 메서드를 기반으로 동기적으로 메시지를 게시할 수 있어요.

비동기 게시를 사용하면 프로듀서가 메시지를 블로킹 큐에 넣고 즉시 반환해요. 그러면 클라이언트 라이브러리가 브로커에 메시지를 백그라운드로 보내요. 큐가 가득 차면(최대 크기 구성 가능) 프로듀서에 전달된 인자에 따라 API 호출 시 프로듀서가 블록되거나 즉시 실패해요.

이 예시는 프로듀서를 사용해 메시지를 게시하는 방법을 보여줘요. 게시 연산은 브로커가 메시지가 성공적으로 게시됐다고 알려줄 때까지 수행돼요. 브로커는 메시지가 성공적으로 게시된 후 메시지 ID를 반환해요.

• Java C++ Go Node.js C# Python

Java로 메시지를 동기적으로 게시하기: MessageId messageId = producer.newMessage().value("my-sync-message").send(); Java로 메시지를 비동기적으로 게시하기: producer.newMessage().value("my-sync-message").sendAsync().thenAccept(messageId -> { log.info("Message ID: {}", messageId); });

Message msg = MessageBuilder()
                    .setContent("my-sync-message")
                    .build();
Result res = producer.send(msg);

 msg := pulsar.ProducerMessage{
     Payload: []byte("my-sync-message"),
 }
 if _, err := producer.send(msg); err != nil {
   log.Fatalf("Could not publish message due to: %v", err)
 }

ProducerMessage 객체의 모든 메서드는 Go API 문서를 참고하세요.

const msg = {
  data: Buffer.from('my-sync-message'),
}
await producer.send(msg);

프로듀서 메시지 객체에 사용할 수 있는 키는 다음과 같아요.

파라미터	설명
data	메시지의 실제 데이터 페이로드.
properties	메시지에 첨부된 애플리케이션별 메타데이터용 Object.
eventTimestamp	메시지와 연관된 타임스탬프.
sequenceId	메시지의 시퀀스 ID.
partitionKey	메시지와 연관된 선택적 키 (토픽 컴팩션 같은 것에 특히 유용).
replicationClusters	이 메시지가 복제되는 클러스터. Pulsar 브로커가 메시지 복제를 자동으로 처리하므로, 브로커 기본값을 덮어쓰고 싶을 때만 이 설정을 바꿔야 해요.
deliverAt	메시지가 전달되는 절대 타임스탬프.
deliverAfter	메시지가 전달되는 상대 지연.

메시지 객체 연산 — Pulsar Node.js 클라이언트에서는 컨슈머(또는 리더)로 메시지 객체를 받을(또는 읽을) 수 있어요. 메시지 객체의 메서드:

메서드	설명	반환 타입
getTopicName()	토픽 이름의 getter 메서드.	String
getProperties()	프로퍼티의 getter 메서드.	Array<Object>
getData()	메시지 데이터의 getter 메서드.	Buffer
getMessageId()	메시지 ID 객체의 getter 메서드.	Object
getPublishTimestamp()	게시 타임스탬프의 getter 메서드.	Number
getEventTimestamp()	이벤트 타임스탬프의 getter 메서드.	Number
getRedeliveryCount()	재전달 횟수의 getter 메서드.	Number
getPartitionKey()	파티션 키의 getter 메서드.	String

메시지 ID 객체 연산 — Pulsar Node.js 클라이언트에서는 메시지 객체에서 메시지 ID 객체를 얻을 수 있어요. 메시지 ID 객체의 메서드:

메서드	설명	반환 타입
serialize()	메시지 ID를 저장용 Buffer로 직렬화.	Buffer
toString()	메시지 ID를 String으로 가져오기.	String

클라이언트는 메시지 ID 객체의 정적 메서드를 갖고 있어요. Pulsar.MessageId.someStaticMethod로 접근할 수 있어요. 정적 메서드:

메서드	설명	반환 타입
earliest()	가장 이른 메시지로 리더를 시작하기 위한 MessageId.
latest()	가장 최근 메시지로 리더를 시작하기 위한 MessageId.

var data = Encoding.UTF8.GetBytes("Hello World");
await producer.Send(data);

producer.send('Hello World'.encode('utf-8'))

메시지 구성 (Configure messages)

Pulsar 메시지의 다양한 속성을 설정할 수 있어요. 이 속성들의 값은 메시지의 메타데이터에 저장돼요.

• Java C++ Go C# Python

producer.newMessage()
              .key("my-key") // Set the message key
              .eventTime(System.currentTimeMillis()) // Set the event time
              .sequenceId(1203) // Set the sequenceId for the deduplication purposes
              .deliverAfter(1, TimeUnit.HOURS) // Delay message delivery for 1 hour
              .property("my-key", "my-value") // Set the customized metadata
              .property("my-other-key", "my-other-value")
              .replicationClusters(
                      Lists.newArrayList("r1", "r2")) // Set the geo-replication clusters for this message.
              .value("content")
              .send();

Java 클라이언트에서는 loadConf를 사용해 메시지 메타데이터를 구성할 수도 있어요. 예시: Map<String, Object> conf = new HashMap<>(); conf.put("key", "my-key"); conf.put("eventTime", System.currentTimeMillis()); producer.newMessage().value("my-message").loadConf(conf).send();

Message msg = MessageBuilder()
                  .setContent("content")
                  .setProperty("my-key", "my-value")
                  .setProperty("my-other-key", "my-other-value")
                  .setDeliverAfter(std::chrono::minutes(3)) // Delay message delivery for 3 minutes
                  .build();
Result res = producer.send(msg);

ID, err := producer.Send(context.Background(), &pulsar.ProducerMessage{
     Payload:      []byte(fmt.Sprintf("content")),
     DeliverAfter: 3 * time.Second, // Delay message delivery for 3 seconds
 })
 if err != nil {
     log.Fatal(err)
 }

var messageId = await producer.NewMessage()
                            .Property("SomeKey", "SomeValue")
                            .Send(data);

producer.send('content'.encode('utf-8'),
              properties={'my-key': 'my-value', 'my-other-key': 'my-other-value'},
              event_timestamp=int(time.time() * 1000))

파티셔닝된 토픽에 메시지 게시 (Publish messages to partitioned topics)

기본적으로 Pulsar 토픽은 단일 브로커가 서비스하므로 토픽의 최대 처리량이 제한돼요. 파티셔닝된 토픽은 여러 브로커에 걸쳐 있을 수 있어 더 높은 처리량을 허용해요.

Pulsar 클라이언트 라이브러리를 사용해 파티셔닝된 토픽에 메시지를 게시할 수 있어요. 파티셔닝된 토픽에 메시지를 게시할 때는 라우팅 모드를 지정해야 해요. 새 프로듀서를 만들 때 라우팅 모드를 지정하지 않으면 라운드 로빈 라우팅 모드가 사용돼요.

내장 메시지 라우터 사용 (Use built-in message router)

라우팅 모드는 각 메시지가 게시될 파티션(내부 토픽)을 결정해요.

다음은 예시예요.

• Java C++ Go Python

Producer<byte[]> producer = pulsarClient.newProducer()
   .topic("my-topic")
   .messageRoutingMode(MessageRoutingMode.SinglePartition)
   .create();

#include "lib/RoundRobinMessageRouter.h" // Make sure include this header file
Producer producer;
Result result = client.createProducer(
   "my-topic",
   ProducerConfiguration().setMessageRouter(std::make_shared<RoundRobinMessageRouter>(
       ProducerConfiguration::BoostHash, true, 1000, 100000, boost::posix_time::seconds(1))),
   producer);

 producer, err := client.CreateProducer(pulsar.ProducerOptions{
     Topic: "my-topic",
     MessageRouter: func(msg *pulsar.ProducerMessage, tm pulsar.TopicMetadata) int {
         fmt.Println("Topic has", tm.NumPartitions(), "partitions. Routing message ", msg, " to partition 2.")
         // always push msg to partition 2
         return 2
     },
 })

 producer = client.create_producer(
     'my-topic',
     message_routing_mode=PartitionsRoutingMode.SinglePartition
 )

커스텀 메시지 라우터 (Customize message router)

• Java C++ Go Python

커스텀 메시지 라우터를 사용하려면 MessageRouter 인터페이스 구현을 제공해야 해요. choosePartition(Message<?>, TopicMetadata) 메서드를 구현하세요(단일 인자 오버로드는 1.22.0부터 deprecated):

public interface MessageRouter extends Serializable {
    int choosePartition(Message<?> msg, TopicMetadata metadata);
}

다음 라우터는 모든 메시지를 파티션 10으로 라우팅해요:

public class AlwaysTenRouter implements MessageRouter {
    @Override
    public int choosePartition(Message<?> msg, TopicMetadata metadata) {
        return 10;
    }
}

그 구현으로 파티셔닝된 토픽에 다음과 같이 메시지를 보낼 수 있어요.

Producer<byte[]> producer = pulsarClient.newProducer()
    .topic("my-topic")
    .messageRouter(new AlwaysTenRouter())
    .create();
producer.send("Partitioned topic message".getBytes());

커스텀 메시지 라우터를 사용하려면 getPartition 메서드 하나를 가진 MessageRoutingPolicy 인터페이스 구현을 제공해야 해요:

class MessageRouter : public MessageRoutingPolicy {
  public:
    MessageRouter() : { }
    int getPartition(const Message& msg, const TopicMetadata& topicMetadata) {
      // The implementation of getPartition
    }
};

다음 라우터는 모든 메시지를 파티션 10으로 라우팅해요:

class MessageRouter : public MessageRoutingPolicy {
  public:
    MessageRouter() { }
    int getPartition(const Message& msg, const TopicMetadata& topicMetadata) {
      return 10;
    }
};

그 구현으로 파티셔닝된 토픽에 아래와 같이 메시지를 보낼 수 있어요.

Producer producer;
Result result = client.createProducer(
    "my-topic",
    ProducerConfiguration().setMessageRouter(std::make_shared<MessageRouter>()),
    producer);
Message msg = MessageBuilder()
    .setContent("content")
    .build();
result = producer.send(msg);

Go 클라이언트에서는 함수를 전달해 커스텀 메시지 라우터를 구성할 수 있어요.

 producer, err := client.CreateProducer(pulsar.ProducerOptions{
     Topic: "my-topic",
     MessageRouter: func(msg *pulsar.ProducerMessage, tm pulsar.TopicMetadata) int {
         fmt.Println("Topic has", tm.NumPartitions(), "partitions. Routing message ", msg, " to partition 10.")
         // always push msg to partition 10
         return 10
     },
 })

Python 클라이언트에서는 함수를 전달해 커스텀 메시지 라우터를 구성할 수 있어요.

 def custom_message_router(msg: pulsar.Message, num_partitions: int):
     # always push msg to partition 10
     return 10

 producer = client.create_producer(
     'my-topic',
     message_router=custom_message_router
 )
 producer.send(b'content')

키를 사용할 때 파티션 선택 (Choose partitions when using a key)

메시지에 키가 있으면 라운드 로빈 라우팅 정책을 대체해요. 다음 Java 예시 코드는 키를 사용할 때 파티션을 선택하는 방법을 설명해요.

// If the message has a key, it supersedes the round robin routing policy
if (msg.hasKey()) {
    return signSafeMod(hash.makeHash(msg.getKey()), topicMetadata.numPartitions());
}

if (isBatchingEnabled) { // if batching is enabled, choose partition on `partitionSwitchMs` boundary.
    long currentMs = clock.millis();
    return signSafeMod(currentMs / partitionSwitchMs + startPtnIdx, topicMetadata.numPartitions());
} else {
    return signSafeMod(PARTITION_INDEX_UPDATER.getAndIncrement(this), topicMetadata.numPartitions());
}

청킹 활성화 (Enable chunking)

메시지 청킹을 사용하면 Pulsar가 대용량 페이로드 메시지를 처리할 수 있는데, 프로듀서 쪽에서 메시지를 청크(chunk)로 분할하고 컨슈머 쪽에서 청크된 메시지를 다시 모아요.

메시지 청킹 기능은 기본적으로 꺼져 있어요. 다음은 프로듀서를 만들 때 메시지 청킹을 활성화하는 예시예요.

• Java C++ Go Python

  Producer<byte[]> producer = client.newProducer()
     .topic(topic)
     .enableChunking(true)
     .enableBatching(false)
     .create();

ProducerConfiguration conf;
conf.setBatchingEnabled(false);
conf.setChunkingEnabled(true);
Producer producer;
client.createProducer("my-topic", conf, producer);

// The message chunking feature is OFF by default.
// By default, a producer chunks the large message based on the max message size (`maxMessageSize`) configured at the broker side (for example, 5MB).
// Client can also configure the max chunked size using the producer configuration `ChunkMaxMessageSize`.
// Note: to enable chunking, you need to disable batching (`DisableBatching=true`) concurrently.
producer, err := client.CreateProducer(pulsar.ProducerOptions{
  Topic:               "my-topic",
  DisableBatching:     true,
  EnableChunking:      true,
})
if err != nil {
	log.Fatal(err)
}
defer producer.Close()

producer = client.create_producer(
        topic,
        chunking_enabled=True
    )

기본적으로 프로듀서는 브로커에 구성된 최대 메시지 크기(maxMessageSize, 예: 5MB)에 따라 대용량 메시지를 청크로 나눠요. 하지만 클라이언트는 프로듀서 구성 chunkMaxMessageSize를 사용해 최대 청크 크기를 구성할 수도 있어요.

청킹을 활성화하려면 배칭(enableBatching = false)을 동시에 비활성화해야 해요.

메시지 인터셉트 (Intercept messages)

ProducerInterceptor는 프로듀서가 받은 메시지가 브로커에 게시되기 전에 가로채서 변형할 수 있어요.

인터페이스에는 세 가지 주요 이벤트가 있어요.

• eligible — 인터셉터를 메시지에 적용할 수 있는지 확인해요.
• beforeSend — 프로듀서가 메시지를 브로커에 보내기 전에 트리거돼요. 이 이벤트에서 메시지를 수정할 수 있어요.
• onSendAcknowledgement — 메시지가 브로커에 ack되거나 전송이 실패했을 때 트리거돼요.

메시지를 인터셉트하려면 Producer를 만들 때 ProducerInterceptor를 하나 또는 여러 개 추가할 수 있어요.

• Java C++

Producer<byte[]> producer = client.newProducer()
     .topic(topic)
     .intercept(new ProducerInterceptor() {
         @Override
         public void close() {
             // release any resources held by the interceptor
         }
         @Override
         public boolean eligible(Message<?> message) {
             return true;  // process all messages
         }
         @Override
         public Message<?> beforeSend(Producer<?> producer, Message<?> message) {
             // user-defined processing logic; return the (possibly modified) message
             return message;
         }
         @Override
         public void onSendAcknowledgement(Producer<?> producer, Message<?> message,
                                           MessageId msgId, Throwable exception) {
             // user-defined processing logic
         }
     })
     .create();

커스텀 인터셉터 구현:

class MyInterceptor : public ProducerInterceptor {
  public:
    MyInterceptor() { }

    Message beforeSend(const Producer& producer, const Message& message) override {
      // Your implementation code
      return message;
    }
    void onSendAcknowledgement(const Producer& producer, Result result, const Message& message, const MessageId& messageID) override {
      // Your implementation code
    }
    void close() override {
      // Your implementation code
    }
};

프로듀서 구성:

ProducerConfiguration conf;
conf.intercept({ std::make_shared<MyInterceptor>(), std::make_shared<MyInterceptor>() });  // You can add multiple interceptors to the same producer
Producer producer;
client.createProducer(topic, conf, producer);

여러 인터셉터는 intercept 메서드에 전달된 순서대로 적용돼요.

암호화 정책 구성 (Configure encryption policies)

Pulsar C# 클라이언트는 네 가지 종류의 암호화 정책을 지원해요.

• EnforceUnencrypted — 항상 암호화되지 않은 연결을 사용해요.
• EnforceEncrypted — 항상 암호화된 연결을 사용해요.
• PreferUnencrypted — 가능하면 암호화되지 않은 연결을 사용해요.
• PreferEncrypted — 가능하면 암호화된 연결을 사용해요.

이 예시는 EnforceUnencrypted 암호화 정책을 설정하는 방법을 보여줘요.

• C#

using DotPulsar;

var client = PulsarClient.Builder()
                      .ConnectionSecurity(EncryptionPolicy.EnforceEncrypted)
                      .Build();

접근 모드 구성 (Configure access mode)

접근 모드를 사용하면 애플리케이션이 토픽에 대한 배타적 프로듀서 접근을 요구해 "단일 작성자(single-writer)" 상황을 만들 수 있어요.

이 예시는 프로듀서 접근 모드를 설정하는 방법을 보여줘요.

• Java C++

이 기능은 Java 클라이언트 2.8.0 이상 버전에서 지원돼요.

Producer<byte[]> producer = client.newProducer()
     .topic(topic)
     .accessMode(ProducerAccessMode.Exclusive)
     .create();

이 기능은 C++ 클라이언트 3.1.0 이상 버전에서 지원돼요.

 Producer producer;
 ProducerConfiguration producerConfiguration;
 producerConfiguration.setAccessMode(ProducerConfiguration::Exclusive);
 client.createProducer(topicName, producerConfiguration, producer);

더 알아보기 (Learn more)

• 컨슈머 사용하기 — 메시지를 소비하는 방법을 알아봐요.
• 리더 사용하기 — 메시지를 읽는 방법을 알아봐요.
• 클라이언트 라이브러리 — 다양한 언어 클라이언트를 살펴봐요.
• 클라이언트 기능 매트릭스 — 언어별 클라이언트 기능 지원을 비교해봐요.