tencent cloud

 pip install 'pulsar-client==3.1.0'

 # Create a client
 client = pulsar.Client(
     authentication=pulsar.AuthenticationToken(
         # Authorized role token
         AUTHENTICATION),
     # Service access address
     service_url=SERVICE_URL)

 # Create a producer
 producer = client.create_producer(
     # Complete path of the topic in the format of `persistent://cluster (tenant) ID/namespace/topic name`, which can be copied from the **Topic** page.
     topic='pulsar-xxx/sdk_python/topic1'
 )

 # Send the message
 producer.send(
     # Message content
     'Hello python client, this is a msg.'.encode('utf-8'),
     # Message parameter
     properties={'k': 'v'},
     # Business key
     partition_key='yourKey'
 )

 # Send the callback in async mode
 def send_callback(send_result, msg_id):
     print('Message published: result:{}  msg_id:{}'.format(send_result, msg_id))

 # Send the message
 producer.send_async(
     # Message content
     'Hello python client, this is a async msg.'.encode('utf-8'),
     # Async callback
     callback=send_callback,
     # Message configuration
     properties={'k': 'v'},
     # Business key
     partition_key='yourKey'
 )

 # Subscribe to the message
 consumer = client.subscribe(
     # Complete path of the topic in the format of `persistent://cluster (tenant) ID/namespace/topic name`, which can be copied from the **Topic** page.
     topic='pulsar-xxx/sdk_python/topic1',
     # Subscription name
     subscription_name='sub_topic1'
 )

 # Obtain the message
 msg = consumer.receive()
 try:
     # Simulate the business processing logic
     print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
     # Return `ack` as the acknowledgement if the consumption is successful
     consumer.acknowledge(msg)
 except:
     # If the consumption fails, the message will be delivered again.
     consumer.negative_acknowledge(msg)