tencent cloud

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

 # 发送消息
 producer.send(
     # 消息内容
     'Hello python client, this is a msg.'.encode('utf-8'),
     # 消息参数
     properties={'k': 'v'},
     # 业务key
     partition_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'},
     # 业务key
     partition_key='yourKey'
 )

 # 订阅消息
 consumer = client.subscribe(
     # topic完整路径，格式为persistent://集群（租户）ID/命名空间/Topic名称，从【Topic管理】处复制
     topic='pulsar-xxx/sdk_python/topic1',
     # 订阅名称
     subscription_name='sub_topic1'
 )

 # 获取消息
 msg = consumer.receive()
 try:
     # 模拟业务
     print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
     # 消费成功，回复ack
     consumer.acknowledge(msg)
 except:
     # 消费失败，消息将会重新投递
     consumer.negative_acknowledge(msg)