Alibaba Cloud Document Center: Find the Documentation for Your Needs

This topic describes how to use the SDK for Python to connect to the default endpoint of a Message Queue for Apache Kafka instance and send and consume messages in a virtual private cloud (VPC).

Prerequisites

Python is installed. For more information, see Download Python.

Note Python 2.7, 3.4, 3.5, 3.6, and 3.7 are supported.
pip is installed. For more information, see Installation.

Install the Python library

Run the following command to install the Python library:

pip install kafka-python

Preparations

Create a Message Queue for Apache Kafka configuration file named setting.py.

kafka_setting = {
    'bootstrap_servers': ["XXX", "XXX", "XXX"],
    'topic_name': 'XXX',
    'consumer_id': 'XXX'
}


Parameter	Description
bootstrap_servers	The default endpoint of the Message Queue for Apache Kafka instance. You can obtain the default endpoint in the Endpoint Information section of the Instance Details page in the Message Queue for Apache Kafka console.
topic_name	The name of the topic. You can obtain the name of the topic on the Topics page in the Message Queue for Apache Kafka console.
consumer_id	The name of the consumer group. Group You can obtain the name of the consumer group on the Groups page in the Message Queue for Apache Kafka console.

Send messages

Create a producer program named aliyun_kafka_producer.py.

#!/usr/bin/env python
# encoding: utf-8

import socket
from kafka import KafkaProducer
from kafka.errors import KafkaError
import setting

conf = setting.kafka_setting

print conf


producer = KafkaProducer(bootstrap_servers=conf['bootstrap_servers'],
                        api_version = (0,10),
                        retries=5)

partitions = producer.partitions_for(conf['topic_name'])
print 'Partitions of the topic: %s' % partitions

try:
    future = producer.send(conf['topic_name'], 'hello aliyun-kafka!')
    future.get()
    print 'send message succeed.'
except KafkaError, e:
    print 'send message failed.'
    print e

Run the following command to run aliyun_kafka_producer.py to send messages:
```
python aliyun_kafka_producer.py
```

Consume messages

Create a consumer program named aliyun_kafka_consumer.py.

#!/usr/bin/env python
# encoding: utf-8

import socket
from kafka import KafkaConsumer
from kafka.errors import KafkaError
import setting

conf = setting.kafka_setting

consumer = KafkaConsumer(bootstrap_servers=conf['bootstrap_servers'],
                        group_id=conf['consumer_id'],
                        api_version = (0,10,2), 
                        session_timeout_ms=25000,
                        max_poll_records=100,
                        fetch_max_bytes=1 * 1024 * 1024)

print 'consumer start to consuming...'
consumer.subscribe((conf['topic_name'], ))
for message in consumer:
    print message.topic, message.offset, message.key, message.value, message.value, message.partition

Run the following command to run aliyun_kafka_consumer.py to consume messages:
```
python aliyun_kafka_consumer.py
```