Pulsar Python client

Python 客户端中生产者、消费者和 Reader 的所有方法都是线程安全的。

关于 pdoc 生成的适用于 Python 客户端的 API 文档，可以参阅。

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

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

可选依赖

为了支持 Pulsar 函数或 Avro 序列化等方面，可以在 pulsar-client 库安装额外的可选组件。

# avro serialization
$ pip install pulsar-client[avro]=='2.9.2'
# functions runtime
$ pip install pulsar-client[functions]=='2.9.2'
# all optional components
$ pip install pulsar-client[all]=='2.9.2'

Installation via PyPi is available for the following Python versions:

Install from source

To install the pulsar-client library by building from source, follow instructions 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

The complete Python API reference is available at .

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'))
client.close()

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.

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()))
        # 确认消息已经成功收到和处理
        consumer.acknowledge(msg)
    except:
        # 消息未被成功处理
client.close()

This example shows how to configure negative acknowledgement.

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)
    print("no more msg")
    pass

读者接口示例

You can use the Pulsar Python API to use the Pulsar reader interface. 下面是一个示例：

# 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()))
    # 无确认操作

多主题订阅

The following is an example.

Declare and validate schema

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()
    b = Integer()
    c = Boolean()

With this simple schema definition, you can create producers, consumers and readers instances that refer to that.

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:

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:
        consumer.negative_acknowledge(msg)

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.

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

字段参数

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):
    a = String()
    sub = MySubRecord()

可以让应用在生产端加密消息并在消费端解密消息。

Configuration

使用 Python 客户端的端到端加密功能，你需要为生产者和消费者配置 publicKeyPath 和 privateKeyPath。

教程

前提条件

Pulsar Python 客户端为 2.7.1或更高版本

步骤

创建公钥和私钥密钥对。

输入

openssl genrsa -out private.pem 2048
openssl rsa -in private.pem -pubout -out public.pem

创建一个生产者用来发送加密消息。

输入

import pulsar
publicKeyPath = "./public.pem"
privateKeyPath = "./private.pem"
crypto_key_reader = pulsar.CryptoKeyReader(publicKeyPath, privateKeyPath)
client = pulsar.Client('pulsar://localhost:6650')
producer = client.create_producer(topic='encryption', encryption_key='encryption', crypto_key_reader=crypto_key_reader)
producer.send('encryption message'.encode('utf8'))
print('sent message')
producer.close()
client.close()

创建消费者接收加密消息。

输入

import pulsar
publicKeyPath = "./public.pem"
privateKeyPath = "./private.pem"
crypto_key_reader = pulsar.CryptoKeyReader(publicKeyPath, privateKeyPath)
client = pulsar.Client('pulsar://localhost:6650')
consumer = client.subscribe(topic='encryption', subscription_name='encryption-sub', crypto_key_reader=crypto_key_reader)
msg = consumer.receive()
print("Received msg '{}' id = '{}'".format(msg.data(), msg.message_id()))
consumer.close()
client.close()

运行消费者接收加密消息。

输入
```
python consumer.py
```
在一个新的终端窗口中，运行生产者来生成加密消息。

输入
```
python producer.py
```
现在你可以看到生产者发送消息，消费者成功收到消息。

输出
消费端：