The Pulsar Python client library is a wrapper over the existing C++ client library and exposes all of the same features. 你可以在 C++ 客户端源码的 python 子目录中找到 Pulsar Python 客户端的相关源码 。

安装

You can install the pulsar-client library either via PyPi, using pip, or by building the library from source.

Installation using pip

To install the pulsar-client library as a pre-built package using the pip package manager:

$ pip install pulsar-client==2.6.1

Installation via PyPi is available for the following Python versions:

10.11 (El Capitan) — 10.12 (Sierra) —
10.13 (High Sierra) — 10.14 (Mojave) | 2.7, 3.7 | | Linux | 2.7, 3.4, 3.5, 3.6, 3.7 |

Installing from source

To install the pulsar-client library by building from source, follow these instructions and compile the Pulsar C++ client library. That will also build the Python binding for the library.

To install the built Python bindings:

$ git clone https://github.com/apache/pulsar
$ cd pulsar/pulsar-client-cpp/python
$ sudo python setup.py install

API 手册：

The complete Python API reference is available at api/python.

示例

Below you’ll find a variety of Python code examples for the pulsar-client library.

生产者示例

This creates a Python producer for the my-topic topic and send 10 messages on that topic:

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()

消费者示例

This creates a consumer with the my-subscription subscription on the my-topic topic, listen for incoming messages, print the content and ID of messages that arrive, and acknowledge each message to the Pulsar broker:

consumer = client.subscribe('my-topic', 'my-subscription')
while True:
    msg = consumer.receive()
    try:
        print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
        # 确认已经成功处理消息
        consumer.acknowledge(msg)
    except:
        # 消息处理失败
        consumer.negative_acknowledge(msg)
client.close()

读者接口示例

You can use the Pulsar Python API to use the Pulsar reader interface. Here’s an example:

# MessageId 取自先前获取的消息
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()))
    # 无确认操作

Schema

Declaring and validating schema

A schema can be declared by passing a class that inherits from pulsar.schema.Record and defines the fields as class variables. 例如:

from pulsar.schema import *
class Example(Record):
    a = String()
    b = Integer()
    c = Boolean()

With this simple schema definition we can then create producers, consumers and readers instances that will be referring to that.

producer = client.create_producer(
                    topic='my-topic',
                    schema=AvroSchema(Example) )
producer.send(Example(a='Hello', b=1))

When the producer is created, the Pulsar broker will validate that the existing topic schema is indeed of “Avro” type and that the format is compatible with the schema definition of the Example class.

If there is a mismatch, the producer creation will raise an exception.

Once a producer is created with a certain schema definition, it will only accept objects that are instances of the declared schema class.

Similarly, for a consumer/reader, the consumer will return an object, instance of the schema record class, rather than the raw bytes:

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:
        # Message failed to be processed
        consumer.negative_acknowledge(msg)

Supported schema types

There are different builtin schema types that can be used in Pulsar. All the definitions are in the pulsar.schema package.

Schema definition reference

The schema definition is done through a class that inherits from pulsar.schema.Record.

This class can have a number of fields which can be of either pulsar.schema.Field type or even another nested Record. All the fields are also specified in the pulsar.schema package. The fields are matching the AVRO fields types.

Additionally, any Python Enum type can be used as a valid field type

字段参数

When adding a field these parameters can be used in the constructor:

Schema 定义示例

简单定义

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

使用枚举

from enum import Enum
class Color(Enum):
    red = 1
    green = 2
    blue = 3
class Example(Record):
    name = String()
    color = Color

复杂类型

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