Skip to content

python-produce-consume.adoc

Latest commit

 

History

History
135 lines (104 loc) · 3.03 KB

python-produce-consume.adoc

File metadata and controls

135 lines (104 loc) · 3.03 KB

Use the Python {pulsar-short} client with {product}

You can use the Python {pulsar-short} client with {product} to produce and consume messages.

For a complete source code example, see the {product} Examples repository.

Prerequisites

  • A supported Python version:

    • For Linux, versions 3.4 to 3.7 are supported

    • For macOS, version 3.7 is supported

  • An {pulsar-reg} topic in {product}

  • A text editor or IDE

Create a project

  1. Create a folder for a new Python project.

  2. In your new directory, install the {pulsar-short} client library with pip:

    mkdir SimpleProducerConsumer && cd SimpleProducerConsumer

touch index.py

pip install pulsar-client==2.10

Write the script

  1. Create an index.py file containing the following code. This code creates a {pulsar-short} client instance with the topic URL and token authentication.

    index.py
    import pulsar
import time

serviceUrl = "<REPLACE_WITH_SERVICE_URL>";
pulsarToken = "<REPLACE_WITH_PULSAR_TOKEN>";

tenantName = "<REPLACE_WITH_TENANT_NAME>";
namespace = "<REPLACE_WITH_NAMESPACE>";
topicName = "<REPLACE_WITH_TOPIC>";

topic = "persistent://{0}/{1}/{2}".format(tenantName, namespace, topicName)

client = pulsar.Client(serviceUrl, authentication=pulsar.AuthenticationToken(pulsarToken))

    This script is intentionally incomplete. Your IDE might show errors until you complete the next steps.

  2. Provide values for the following variables:

    ROOT:partial$client-variables-table.adoc

  3. Use the client to create a producer:

    index.py
    producer = client.create_producer(topic)

  4. Send a message:

    index.py
    producer.send('Hello World'.encode('utf-8'))

  5. Use the Python client instance to create a consumer subscription to the same topic that you sent a message to:

    index.py
    consumer = client.subscribe(topic, 'my-subscription')

  6. Iterate through messages and write their data:

    index.py
    waitingForMsg = True
while waitingForMsg:
    try:
        msg = consumer.receive(2000)
        print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))

        # Acknowledging the message to remove from message backlog
        consumer.acknowledge(msg)

        waitingForMsg = False
    except:
        print("Still waiting for a message...");

    time.sleep(1)

  7. Clean up:

    client.close()

Run the script

In your Python project directory, run the script:

python3 index.py

The output includes a lot of logs. Received message confirms that the script succeeded:

Received message 'Hello World' id='(422529,5,-1,0)'

Next steps

  • developing:configure-pulsar-env.adoc

  • developing:astream-functions.adoc

  • streaming-learning:pulsar-io:connectors/index.adoc