RabbitMQ Source 커넥터

RabbitMQ Source 커넥터

RabbitMQ source 커넥터는 RabbitMQ 클러스터에서 메시지를 받아 Pulsar 토픽에 메시지를 써 넣는 커넥터예요. RabbitMQ의 메시지를 Pulsar로 가져올 때 사용해요.

참고: 모든 Pulsar 커넥터는 download page에서 내려받을 수 있어요.

출처: 문서

본문

구성 (Configuration)

RabbitMQ source 커넥터의 구성에는 다음과 같은 프로퍼티가 있어요.

프로퍼티 (Property)

예제 (Example)

RabbitMQ source 커넥터를 사용하기 전에 다음 방법 중 하나로 구성 파일을 만들어야 해요.

• JSON

{
   "configs": {
      "host": "localhost",
      "port": "5672",
      "virtualHost": "/",
      "username": "guest",
      "password": "guest",
      "queueName": "test-queue",
      "connectionName": "test-connection",
      "requestedChannelMax": "0",
      "requestedFrameMax": "0",
      "connectionTimeout": "60000",
      "handshakeTimeout": "10000",
      "requestedHeartbeat": "60",
      "prefetchCount": "0",
      "prefetchGlobal": "false",
      "passive": "false"
   }
}

• YAML

configs:
    host: "localhost"
    port: 5672
    virtualHost: "/"
    username: "guest"
    password: "guest"
    queueName: "test-queue"
    connectionName: "test-connection"
    requestedChannelMax: 0
    requestedFrameMax: 0
    connectionTimeout: 60000
    handshakeTimeout: 10000
    requestedHeartbeat: 60
    prefetchCount: 0
    prefetchGlobal: "false"
    passive: "false"

사용법 (Usage)

Standalone 모드 (Standalone mode)

이 예시는 standalone 모드에서 RabbitMQ source 커넥터를 사용해 RabbitMQ에서 데이터를 가져와 Pulsar 토픽에 쓰는 방법을 설명해요.

전제 조건 (Prerequisites)

• 큐에 몇몇 히스토리 메시지가 있는 RabbitMQ 서버가 있어야 해요.

단계 (Steps)

• Pulsar 패키지를 받아 standalone 모드로 Pulsar를 시작해요.

curl -LO "https://www.apache.org/dyn/closer.lua/pulsar/pulsar-5.0.0-M2/apache-pulsar-5.0.0-M2-bin.tar.gz?action=download"
tar xvfz apache-pulsar-5.0.0-M2-bin.tar.gz
cd apache-pulsar-5.0.0-M2
bin/pulsar standalone

• Pulsar 버전에 해당하는 nar 패키지를 다운로드하고 다음 파일을 Pulsar 디렉터리에 복사해요.

cd connectors
curl -LO "https://www.apache.org/dyn/closer.lua/pulsar/pulsar-5.0.0-M2/connectors/pulsar-io-rabbitmq-5.0.0-M2.nar?action=download"

• 기본적으로 durable 구독이 하나도 없는 토픽에 발행된 메시지는 삭제 준비 상태로 자동 표시돼요. 이를 막기 위해 네임스페이스 수준에서 보존 정책(retention policy)을 설정할 수 있어요.

./bin/pulsar-admin namespaces set-retention -s 100M -t 3d public/default

• 이름이 rabbitmq-source-queue-name.yaml인 구성 파일을 준비해요.

 configs:
   host: "localhost"
   port: 5672
   virtualHost: "/"
   username: "guest"
   password: "guest"
   queueName: "test-queue"
   connectionName: "test-connection"
   requestedChannelMax: 0
   requestedFrameMax: 0
   connectionTimeout: 60000
   handshakeTimeout: 10000
   requestedHeartbeat: 60
   prefetchCount: 0
   prefetchGlobal: "false"
   passive: "false"

구성 파일을 Pulsar의 conf 디렉터리에 복사해요.

cp rabbitmq-source-queue-name.yaml ./conf

• 새 터미널 창을 열고 커넥터를 로컬 실행 모드로 시작해요.

./bin/pulsar-admin source localrun \
 --source-config-file $PWD/conf/rabbitmq-source-queue-name.yaml \
 --archive $PWD/connectors/pulsar-io-rabbitmq-5.0.0-M2.nar \
 --name rabbitmq-source \
 --destination-topic-name pulsar-rabbitmq-test-topic \
 --broker-service-url pulsar://{ip}:{port}

• 새 터미널 창을 열고 토픽이 자동으로 생성됐는지 확인해요.

./bin/pulsar-admin topics list public/default \

이 토픽은 다음과 같이 자동 생성돼요.

persistent://public/default/pulsar-rabbitmq-test-topic-partition-0

• 이 토픽을 소비해요.

./bin/pulsar-client consume persistent://public/default/pulsar-rabbitmq-test-topic-partition-0 -s s1 -n 0 -p Earliest

컨슈머 터미널 창에 다음 정보가 나타나요.

   ----- got message -----
   key:[quick.orange.pulsar], properties:[], content:message-topic-O(range) 0
   ----- got message -----
   key:[quick.orange.pulsar], properties:[], content:message-topic-O(range) 1
   ... ...

더 알아보기 (Learn more)

• RabbitMQ 커넥터 개요
• RabbitMQ sink 커넥터
• Pulsar IO 커넥터 개요
• 커넥터 관리 CLI (pulsar-admin)