Python 클라이언트 사용

Python 클라이언트 사용 (Use a Python client)

Pulsar Python 클라이언트로 프로듀서, 컨슈머, 리더를 만들어 메시지를 주고받는 방법을 살펴볼게요. 기본 사용법부터 스키마를 정의하고 적용하는 부분까지 차근차근 정리했어요. 파이썬답게 간결한 API라 금방 익숙해질 거예요.

출처: 문서

본문

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

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

import pulsar

client = pulsar.Client('pulsar://localhost:6650')

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

for i in range(10):
    producer.send(('Hello-%d' % i).encode('utf-8'))

client.close()

컨슈머 만들기 (Create a consumer)

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

import pulsar

client = pulsar.Client('pulsar://localhost:6650')

consumer = client.subscribe('my-topic', 'my-subscription')

while True:
    msg = consumer.receive()
    try:
        print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
        # Acknowledge successful processing of the message
        consumer.acknowledge(msg)
    except Exception:
        # Message failed to be processed
        consumer.negative_acknowledge(msg)

client.close()

이 예시는 부정 ack(negative acknowledgment)를 구성하는 방법을 보여줘요.

from pulsar import Client, schema

client = Client('pulsar://localhost:6650')
consumer = client.subscribe('negative_acks','test',schema=schema.StringSchema())

producer = client.create_producer('negative_acks',schema=schema.StringSchema())

for i in range(10):
    print('send msg "hello-%d"' % i)
    producer.send_async('hello-%d' % i, callback=None)

producer.flush()

for i in range(10):
    msg = consumer.receive()
    consumer.negative_acknowledge(msg)
    print('receive and nack msg "%s"' % msg.data())

for i in range(10):
    msg = consumer.receive()
    consumer.acknowledge(msg)
    print('receive and ack msg "%s"' % msg.data())

try:
    # No more messages expected
    msg = consumer.receive(100)
except:
    print("no more msg")
    pass

리더 만들기 (Create a reader)

Pulsar Python API를 사용해 Pulsar 리더 인터페이스를 사용할 수 있어요. 예시는 다음과 같아요.

# MessageId taken from a previously fetched message
msg_id = msg.message_id()

reader = client.create_reader('my-topic', msg_id)

while True:
    msg = reader.read_next()
    print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
    # No acknowledgment

스키마로 작업하기 (Work with schema)

지원되는 스키마 타입 (Supported schema types)

Pulsar에는 다양한 내장 스키마 타입을 사용할 수 있어요. 모든 정의는 pulsar.schema 패키지에 있어요.

스키마 정의 참조 (Schema definition reference)

스키마 정의는 pulsar.schema.Record를 상속하는 클래스를 통해 이뤄져요. 이 클래스에는 pulsar.schema.Field 타입이거나 또 다른 중첩 Record일 수 있는 여러 필드가 있어요. 모든 필드는 pulsar.schema 패키지에 명시되어 있고, 필드는 AVRO 필드 타입과 일치해요.

추가로, 모든 Python Enum 타입을 유효한 필드 타입으로 사용할 수 있어요.

필드 파라미터 (Fields parameters)

필드를 추가할 때 생성자에서 다음 파라미터를 사용할 수 있어요.

스키마 정의 예시 (Schema definition examples)

간단한 정의 (Simple definition)

class Example(Record):
    a = String()
    b = Integer()
    c = Array(String())
    i = Map(String())

Enum 사용 (Using enums)

from enum import Enum

class Color(Enum):
    red = 1
    green = 2
    blue = 3

class Example(Record):
    name = String()
    color = Color

복합 타입 (Complex types)

class MySubRecord(Record):
    x = Integer()
    y = Long()
    z = String()

class Example(Record):
    a = String()
    sub = MySubRecord()

Avro 스키마의 네임스페이스 설정 (Set namespace for Avro schema)

특수 필드 _avro_namespace를 사용해 Avro Record 스키마의 네임스페이스를 설정해요.

class NamespaceDemo(Record):
   _avro_namespace = 'xxx.xxx.xxx'
   x = String()
   y = Integer()

스키마 정의는 다음과 같아요.

{
  "name": "NamespaceDemo",
  "namespace": "xxx.xxx.xxx",
  "type": "record",
  "fields": [
    {"name": "x", "type": ["null", "string"]},
    {"name": "y", "type": ["null", "int"]}
  ]
}

스키마 선언 및 검증 (Declare and validate schema)

프로듀서가 만들어지기 전에 Pulsar 브로커는 기존 토픽 스키마가 올바른 타입이고 그 형식이 클래스의 스키마 정의와 호환되는지 검증해요. 토픽 스키마의 형식이 스키마 정의와 호환되지 않으면 프로듀서 생성에서 예외가 발생해요.

특정 스키마 정의로 프로듀서가 한 번 만들어지면, 선언된 스키마 클래스의 인스턴스인 객체만 받아들여요.

마찬가지로, 컨슈머나 리더의 경우 컨슈머는 원시 바이트가 아니라 스키마 레코드 클래스의 인스턴스인 객체를 반환해요.

예시:

consumer = client.subscribe(
                  topic='my-topic',
                  subscription_name='my-subscription',
                  schema=AvroSchema(Example) )

while True:
    msg = consumer.receive()
    ex = msg.value()
    try:
        print("Received message a={} b={} c={}".format(ex.a, ex.b, ex.c))
        # Acknowledge successful processing of the message
        consumer.acknowledge(msg)
    except Exception:
        # Message failed to be processed
        consumer.negative_acknowledge(msg)

더 많은 코드 예시는 Schema - Get started를 참고하세요.

더 알아보기 (Learn more)

• Python 클라이언트 설정 — 설치 방법을 알아봐요.
• Python 클라이언트 초기화 — 클라이언트를 만드는 방법을 살펴봐요.
• Python 클라이언트 — Python 클라이언트 개요를 확인해요.
• Python 클라이언트 API 문서 — 구성 파라미터를 자세히 확인해봐요.
• 스키마 시작하기 — 스키마를 정의하고 쓰는 방법을 알아봐요.
• 프로듀서 — 프로듀서 개념을 자세히 알아봐요.