pulsar_client_quickstart_produce_consume_message.py

1import pulsar
2
3client = pulsar.Client('pulsar://localhost:6650')
4
5producer = client.create_producer('my-topic')
6producer.send('Hello Pulsar!'.encode('utf-8'))
7
8consumer = client.subscribe('my-topic', 'my-subscription')
9
10while True:
11    msg = consumer.receive()
12    try:
13        print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
14        # Acknowledge successful processing of the message
15        consumer.acknowledge(msg)
16    except Exception:
17        # Message failed to be processed
18        consumer.negative_acknowledge(msg)
19
20client.close()