Node.js 클라이언트 사용

Node.js 클라이언트 사용 (Use a Node.js client)

Pulsar Node.js 클라이언트로 프로듀서, 컨슈머, 리더를 실제로 만들어 메시지를 주고받는 방법을 살펴볼게요. 각 개체를 만들 때 구성 객체를 넘기고, 사용 가능한 연산 메서드들을 정리했어요. 마지막에는 리더가 컨슈머와 어떻게 다른지도 확인할 수 있어요.

출처: 문서

본문

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

프로듀서 구성 객체를 사용해 Node.js 프로듀서를 구성할 수 있어요.

예시는 다음과 같아요.

const producer = await client.createProducer({
  topic: 'my-topic', // or 'my-tenant/my-namespace/my-topic' to specify topic's tenant and namespace
});

await producer.send({
  data: Buffer.from("Hello, Pulsar"),
});

await producer.close();

새 Pulsar 프로듀서를 만들면 연산이 Promise 객체를 반환하고, executor 함수를 통해 프로듀서 인스턴스나 오류를 얻어요. 위 예시에서는 executor 함수 대신 await 연산자를 사용했어요.

프로듀서 연산 (Producer operations)

Pulsar Node.js 프로듀서에는 다음 메서드가 있어요.

프로듀서 예시 (Producer example)

이 예시는 my-topic 토픽에 Node.js 프로듀서를 만들고 10개의 메시지를 그 토픽에 보내요.

const Pulsar = require('pulsar-client');

(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });

  // Create a producer
  const producer = await client.createProducer({
    topic: 'my-topic',
  });

  // Send messages
  for (let i = 0; i < 10; i += 1) {
    const msg = `my-message-${i}`;
    producer.send({
      data: Buffer.from(msg),
    });
    console.log(`Sent message: ${msg}`);
  }
  await producer.flush();

  await producer.close();
  await client.close();
})();

컨슈머 만들기 (Create a consumer)

컨슈머 구성 객체를 사용해 Node.js 컨슈머를 구성할 수 있어요.

예시는 다음과 같아요.

const consumer = await client.subscribe({
  topic: 'my-topic',
  subscription: 'my-subscription',
});

const msg = await consumer.receive();
console.log(msg.getData().toString());
consumer.acknowledge(msg);

await consumer.close();

새 Pulsar 컨슈머를 만들면 연산이 Promise 객체를 반환하고, executor 함수를 통해 컨슈머 인스턴스나 오류를 얻어요. 이 예시에서는 executor 함수 대신 await 연산자를 사용했어요.

컨슈머 연산 (Consumer operations)

Pulsar Node.js 컨슈머에는 다음 메서드가 있어요.

컨슈머 예시 (Consumer example)

이 예시는 my-topic 토픽에 my-subscription 서브스크립션으로 Node.js 컨슈머를 만들고, 메시지를 받아 도착하는 내용을 출력한 다음 각 메시지를 Pulsar 브로커에 10번 ack해요.

const Pulsar = require('pulsar-client');

(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });

  // Create a consumer
  const consumer = await client.subscribe({
    topic: 'my-topic',
    subscription: 'my-subscription',
    subscriptionType: 'Exclusive',
  });

  // Receive messages
  for (let i = 0; i < 10; i += 1) {
    const msg = await consumer.receive();
    console.log(msg.getData().toString());
    consumer.acknowledge(msg);
  }

  await consumer.close();
  await client.close();
})();

대신 컨슈머를 리스너와 함께 만들어 메시지를 처리할 수도 있어요.

// Create a consumer
const consumer = await client.subscribe({
  topic: 'my-topic',
  subscription: 'my-subscription',
  subscriptionType: 'Exclusive',
  listener: (msg, msgConsumer) => {
    console.log(msg.getData().toString());
    msgConsumer.acknowledge(msg);
  },
});

Pulsar Node.js 클라이언트는 AsyncWorker를 사용해요. 컨슈머/프로듀서 만들기, 메시지 수신/송신 같은 비동기 연산은 워커 스레드에서 수행돼요. 이 연산들이 완료될 때까지 워커 스레드는 블록돼요.

기본적으로 워커 스레드가 4개뿐이라 호출된 메서드가 끝나지 않을 수 있어요. 이 상황을 피하려면 UV_THREADPOOL_SIZE를 설정해 워커 스레드 수를 늘리거나, receive()를 여러 번 호출하는 대신 리스너를 정의하세요.

리더 만들기 (Create a reader)

Pulsar 리더는 스트림에서 어떤 메시지부터 시작할지 명시적으로 지정해야 한다는 점에서 컨슈머와 달라요(반면 컨슈머는 자동으로 가장 최근의 unacked 메시지부터 시작해요). 리더 구성 객체를 사용해 Node.js 리더를 구성할 수 있어요.

예시는 다음과 같아요.

const reader = await client.createReader({
  topic: 'my-topic',
  startMessageId: Pulsar.MessageId.earliest(),
});

const msg = await reader.readNext();
console.log(msg.getData().toString());

await reader.close();

리더 연산 (Reader operations)

Pulsar Node.js 리더에는 다음 메서드가 있어요.

리더 예시 (Reader example)

이 예시는 my-topic 토픽으로 Node.js 리더를 만들고, 메시지를 읽어 도착하는 내용을 10번 출력해요.

const Pulsar = require('pulsar-client');

(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
    operationTimeoutSeconds: 30,
  });

  // Create a reader
  const reader = await client.createReader({
    topic: 'my-topic',
    startMessageId: Pulsar.MessageId.earliest(),
  });

  // read messages
  for (let i = 0; i < 10; i += 1) {
    const msg = await reader.readNext();
    console.log(msg.getData().toString());
  }

  await reader.close();
  await client.close();
})();

더 알아보기 (Learn more)

• Node.js 클라이언트 설정 — 설치 방법을 알아봐요.
• Node.js 클라이언트 초기화 — 클라이언트를 만드는 방법을 살펴봐요.
• Node.js 클라이언트 구성 — 사용 가능한 구성 파라미터를 확인해봐요.
• Node.js 클라이언트 — Node.js 클라이언트 개요를 확인해요.
• 프로듀서 — 프로듀서 개념을 자세히 알아봐요.
• 컨슈머 — 컨슈머 개념을 자세히 알아봐요.