pulsar_client_quickstart_produce_consume_message.py

1import pulsar
2
3# Create a pulsar client by supplying ip address and port
4client = pulsar.Client('pulsar://localhost:6650')
5
6# Create a producer on a specific topic
7producer = client.create_producer('my-topic')
8
9# Send a message to the topic
10producer.send('Hello Pulsar!'.encode('utf-8'))
11
12# Create a consumer on the same topic with a subscription name
13consumer = client.subscribe('my-topic', 'my-subscription')
14
15# Receive and acknowledge the message
16msg = consumer.receive()
17try:
18    print("Received message: '%s'" % msg.data().decode('utf-8'))
19    # Acknowledge successful processing of the message
20    consumer.acknowledge(msg)
21except Exception:
22    # Message failed to be processed
23    consumer.negative_acknowledge(msg)
24
25# Close the consumer, producer, and client
26consumer.close()
27producer.close()
28client.close()