Docker에서 Standalone Pulsar 설정하기

Docker에서 Standalone Pulsar 설정하기

로컬 개발과 테스트를 위해 Docker 컨테이너 안에서 standalone 모드로 Pulsar를 실행할 수 있어요. 개발용이므로 한 대의 머신에서 모든 구성 요소를 가볍게 돌리고 싶을 때 아주 유용하죠. 이번에는 Docker로 Pulsar를 띄우고 메시지를 주고받는 것까지 빠르게 해볼게요.

출처: 문서

본문

사전 준비 (Prerequisites)

  • Docker (버전 20.10+ 권장)
  • 사용 가능한 RAM 4GB 이상
  • 여유 디스크 공간 5GB 이상

Docker에서 Pulsar 시작하기 (Start Pulsar in Docker)

  • macOS, Linux, Windows용:
docker run -it -p 6650:6650  -p 8080:8080 --mount source=pulsardata,target=/pulsar/data --mount source=pulsarconf,target=/pulsar/conf apachepulsar/pulsar:5.0.0-M2 bin/pulsar standalone

이 명령에 대해 알아둘 점이 몇 가지 있어요.

  • 데이터, 메타데이터, 구성은 Docker 볼륨에 영속화되어 컨테이너를 재시작할 때마다 "새로 시작"하지 않아요. 볼륨에 대한 자세한 내용은 docker volume inspect <sourcename>을 사용할 수 있어요.
  • Windows의 Docker라면 Linux 컨테이너를 사용하도록 구성했는지 확인해요.
  • docker 컨테이너는 기본적으로 UID 10000과 GID 0으로 실행돼요. 마운트된 볼륨이 UID 10000 또는 GID 0에 쓰기 권한을 주도록 확인해야 해요. UID 10000은 임의의 값이므로, root 그룹(GID 0)에 이 마운트에 대한 쓰기 권한을 주는 것을 권장해요.

Pulsar가 성공적으로 시작되면 다음과 같은 INFO 레벨 로그 메시지를 볼 수 있어요.

08:18:30.970 [main] INFO  org.apache.pulsar.broker.web.WebService - HTTP Service started at http://0.0.0.0:8080...
07:53:37.322 [main] INFO  org.apache.pulsar.broker.PulsarService - messaging service is ready, bootstrap service port = 8080, broker url= pulsar://localhost:6650, cluster=standalone, configs=org.apache.pulsar.broker.ServiceConfiguration@98b63c1...

tip 로컬 standalone 클러스터를 시작하면 public/default

네임스페이스가 자동으로 생성되며 개발 목적으로 사용돼요. 모든 Pulsar 토픽은 네임스페이스 안에서 관리돼요. 자세한 내용은 Topics 문서를 참고해요.

Docker에서 Pulsar 사용하기 (Use Pulsar in Docker)

Pulsar는 Java, Go, Python, C++ 클라이언트 라이브러리를 제공해요. 로컬 standalone 클러스터를 실행 중이라면 다음 루트 URL 중 하나로 클러스터와 상호작용할 수 있어요.

  • pulsar://localhost:6650
  • http://localhost:8080

다음 예제는 Python 클라이언트 API를 사용해 Pulsar를 빠르게 시작할 수 있게 안내해요.

PyPI에서 Pulsar Python 클라이언트 라이브러리를 바로 설치해요.

pip install pulsar-client

메시지 소비 (Consume a message)

컨슈머를 만들고 토픽에 구독해요.

import pulsar
client = pulsar.Client('pulsar://localhost:6650')
consumer = client.subscribe('my-topic',
                            subscription_name='my-sub')
while True:
    msg = consumer.receive()
    print("Received message: '%s'" % msg.data())
    consumer.acknowledge(msg)
client.close()

메시지 생성 (Produce a message)

이제 프로듀서를 시작해 테스트 메시지를 몇 개 보내요.

import pulsar
client = pulsar.Client('pulsar://localhost:6650')
producer = client.create_producer('my-topic')
for i in range(10):
    producer.send(('hello-pulsar-%d' % i).encode('utf-8'))
client.close()

토픽 통계 얻기 (Get the topic statistics)

Pulsar에서는 REST, Java, 또는 커맨드라인 도구로 시스템의 모든 측면을 제어할 수 있어요. API에 대한 자세한 내용은 Admin API 개요를 참고해요.

가장 간단한 예로, curl을 사용해 특정 토픽의 통계를 조회할 수 있어요.

curl http://localhost:8080/admin/v2/persistent/public/default/my-topic/stats | python -m json.tool

출력은 대략 다음과 같아요.

{
    "msgRateIn": 0.0,
    "msgThroughputIn": 0.0,
    "msgRateOut": 1.8332950480217471,
    "msgThroughputOut": 91.33142602871978,
    "bytesInCounter": 7097,
    "msgInCounter": 143,
    "bytesOutCounter": 6607,
    "msgOutCounter": 133,
    "averageMsgSize": 0.0,
    "msgChunkPublished": false,
    "storageSize": 7097,
    "backlogSize": 0,
    "offloadedStorageSize": 0,
    "publishers": [
        {
            "accessMode": "Shared",
            "msgRateIn": 0.0,
            "msgThroughputIn": 0.0,
            "averageMsgSize": 0.0,
            "chunkedMessageRate": 0.0,
            "producerId": 0,
            "metadata": {},
            "address": "/127.0.0.1:35604",
            "connectedSince": "2021-07-04T09:05:43.04788Z",
            "clientVersion": "2.8.0",
            "producerName": "standalone-2-5"
        }
    ],
    "waitingPublishers": 0,
    "subscriptions": {
        "my-sub": {
            "msgRateOut": 1.8332950480217471,
            "msgThroughputOut": 91.33142602871978,
            "bytesOutCounter": 6607,
            "msgOutCounter": 133,
            "msgRateRedeliver": 0.0,
            "chunkedMessageRate": 0,
            "msgBacklog": 0,
            "backlogSize": 0,
            "msgBacklogNoDelayed": 0,
            "blockedSubscriptionOnUnackedMsgs": false,
            "msgDelayed": 0,
            "unackedMessages": 0,
            "type": "Exclusive",
            "activeConsumerName": "3c544f1daa",
            "msgRateExpired": 0.0,
            "totalMsgExpired": 0,
            "lastExpireTimestamp": 0,
            "lastConsumedFlowTimestamp": 1625389101290,
            "lastConsumedTimestamp": 1625389546070,
            "lastAckedTimestamp": 1625389546162,
            "lastMarkDeleteAdvancedTimestamp": 1625389546163,
            "consumers": [
                {
                    "msgRateOut": 1.8332950480217471,
                    "msgThroughputOut": 91.33142602871978,
                    "bytesOutCounter": 6607,
                    "msgOutCounter": 133,
                    "msgRateRedeliv...

더 알아보기 (Learn more)