All the methods in producer, consumer, and reader of a Python client are thread-safe.

pdoc-generated API docs for the Python client are available .

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

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

Installation via PyPi is available for the following Python versions:

Install from source

To install the pulsar-client library by building from source, follow and compile the Pulsar C++ client library. That builds 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

You can find a variety of Python code examples for the pulsar-client library.

生产者示例

The following example creates a Python producer for the my-topic topic and sends 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'))

The following example creates a consumer with the my-subscription subscription name on the my-topic topic, receives incoming messages, prints the content and ID of messages that arrive, and acknowledges 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:

多主题订阅

In addition to subscribing a consumer to a single Pulsar topic, you can also subscribe to multiple topics simultaneously. To use multi-topic subscriptions, you can supply a regular expression (regex) or a List of topics. 如果通过 regex 选择主题, 则所有主题都必须位于同一Pulsar命名空间中。

The following is an example.

import re
consumer = client.subscribe(re.compile('persistent://public/default/topic-*'), 'my-subscription')
while True:
    msg = consumer.receive()
        print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
        # Acknowledge successful processing of the message
        consumer.acknowledge(msg)
    except:
        # Message failed to be processed
        consumer.negative_acknowledge(msg)
client.close()

You can declare a schema 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()
    c = Boolean()

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

After creating the producer, the Pulsar broker validates 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, an exception occurs in the producer creation.

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:

Supported schema types

You can use different builtin schema types 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 has a number of fields which can be of either pulsar.schema.Field type or another nested Record. All the fields are specified in the pulsar.schema package. The fields are matching the AVRO fields types.

字段参数

When adding a field, you can use these parameters 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):
    sub = MySubRecord()