pip install 'pulsar-client==3.1.0'
# 创建客户端client = pulsar.Client(authentication=pulsar.AuthenticationToken(# 已授权角色密钥AUTHENTICATION),# 服务接入地址service_url=SERVICE_URL)
# 创建生产者producer = client.create_producer(# topic完整路径,格式为persistent://集群(租户)ID/命名空间/Topic名称,从【Topic管理】处复制topic='pulsar-xxx/sdk_python/topic1')
persistent://clusterid/namespace/Topic,clusterid/namespace/topic 的部分可以从控制台上 Topic管理 页面直接复制。# 发送消息producer.send(# 消息内容'Hello python client, this is a msg.'.encode('utf-8'),# 消息参数properties={'k': 'v'},# 业务keypartition_key='yourKey')
# 异步发送回调def send_callback(send_result, msg_id):print('Message published: result:{} msg_id:{}'.format(send_result, msg_id))# 发送消息producer.send_async(# 消息内容'Hello python client, this is a async msg.'.encode('utf-8'),# 异步回调callback=send_callback,# 消息配置properties={'k': 'v'},# 业务keypartition_key='yourKey')
# 订阅消息consumer = client.subscribe(# topic完整路径,格式为persistent://集群(租户)ID/命名空间/Topic名称,从【Topic管理】处复制topic='pulsar-xxx/sdk_python/topic1',# 订阅名称subscription_name='sub_topic1')
persistent://clusterid/namespace/Topic,clusterid/namespace/topic 的部分可以从控制台上 Topic管理 页面直接复制。
# 获取消息msg = consumer.receive()try:# 模拟业务print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))# 消费成功,回复ackconsumer.acknowledge(msg)except:# 消费失败,消息将会重新投递consumer.negative_acknowledge(msg)

文档反馈