阿里云kafka使用记录（python版本）

kafka端

 

consumer vpc版代码

 

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

# context.check_hostname = True

consumer = KafkaConsumer(bootstrap_servers=[\'192.168.xx.xx:9092\'],
                        group_id=\'xx\',
                        api_version = (0,10)
                        )

print(\'consumer start to consuming...\')
consumer.subscribe((\'xx\',))
for message in consumer:
    print(message.topic)
    print(message.offset)
    print(message.key)
    print(message.value)
    print(message.partition)

 

producer vpc版代码

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

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

producer = KafkaProducer(bootstrap_servers=[\'192.168.xx.xx:9092\'],
                        api_version = (0,10),
                        retries=5)

partitions = producer.partitions_for(\'xx\')
print(\'Topic下分区: %s\' % partitions)

try:
    future = producer.send(topic=\'xx\', value=b\'hello aliyun-kafka!\')
    future.get()
    print(\'send message succeed.\')
except KafkaError as e:
    print(\'send message failed.\')
    print(e)

consumer公网版代码

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


context = ssl.create_default_context()
context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
context.verify_mode = ssl.CERT_REQUIRED
# context.check_hostname = True
context.load_verify_locations("/tmp/ca-cert")

consumer = KafkaConsumer(bootstrap_servers=[\'kafka-ons-internet.aliyun.com:8080\'],
                        group_id=\'xxx\',
                        sasl_mechanism="PLAIN",
                        ssl_context=context,
                        security_protocol=\'SASL_SSL\',
                        api_version = (0,10),
                        sasl_plain_username=\'xxx\',
                        sasl_plain_password=\'1234567890\')

print(\'consumer start to consuming...\')
consumer.subscribe((\'xxx\', ))
for message in consumer:
    print(message.topic)
    print(message.offset)
    print(message.value)
    break

 

 

producer 公网版代码

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

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

context = ssl.create_default_context()
context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
context.verify_mode = ssl.CERT_REQUIRED
# context.check_hostname = True
context.load_verify_locations("/tmp/ca-cert")
#这个文件参考https://github.com/AliwareMQ/aliware-kafka-demos/tree/master/kafka-python-demo

producer = KafkaProducer(bootstrap_servers=[\'kafka-ons-internet.aliyun.com:8080\'],
                        sasl_mechanism="PLAIN",
                        ssl_context=context,
                        security_protocol=\'SASL_SSL\',
                        api_version = (0,10),
                        retries=5,
                        sasl_plain_username=\'xx\',
                        sasl_plain_password=\'1234567890\'#注意是access-key的最后十位)

partitions = producer.partitions_for(\'xxx\')
print (\'Topic下分区: %s\' % partitions)

try:
    future = producer.send(\'xxx\', b\'hello aliyun-kafka!\')
    future.get()
    print(\'send message succeed.\')
except KafkaError as e:
    print(\'send message failed.\')
    print(e)

 

 

 

 

从阿里云控台获得连接信息