Demo Kết nối Kafka Bizfly Cloud với ngôn ngữ Python

KAFKA_CONFIG = {  
    'bootstrap.servers': 'your-kafka-bootstrap.bizflycloud.vn:9093',  
    'security.protocol': ' SASL_PLAINTEXT',  
    'sasl.mechanism': 'SCRAM-SHA-256,  
    'sasl.username': 'your-username',  
    'sasl.password': 'your-password'  
    }  
TOPIC_NAME = 'your-topic-name'

from confluent_kafka import Producer

conf = {
    'bootstrap.servers': 'IP-Kafka',  # Thay bang dia chi Kafka thuc te
    'security.protocol': 'SASL_PLAINTEXT',
    'sasl.mechanism': 'SCRAM-SHA-256',
    'sasl.username': 'your-username',
    'sasl.password': 'your-password',
}

producer = Producer(conf)

def delivery_report(err, msg):
    if err is not None:
        print(f"Loi gui: {err}")
    else:
        print(f"Gui thanh cong den {msg.topic()} [{msg.partition()}]")

# Gui mot tin nhan don gian
producer.produce('your-topic', key='key1', value='Hello Kafka', callback=delivery_report)
producer.flush()

from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'IP-Kafka',  # Thay bang dia chi Kafka thuc te
    'security.protocol': 'SASL_PLAINTEXT',
    'sasl.mechanism': 'SCRAM-SHA-256',
    'sasl.username': 'your-username',
    'sasl.password': 'your-password', 
    'group.id': 'demo-group',  
    'auto.offset.reset': 'earliest'  
}

consumer = Consumer(conf)  
consumer.subscribe([TOPIC_NAME])

print("🚦 Listening for messages...\nPress Ctrl+C to exit.\n")  
try:  
    while True:  
        msg = consumer.poll(1.0)  
        if msg is None:  
            continue  
        if msg.error():  
            print(f"❌ Consumer error: {msg.error()}")  
        else:  
            print(f"📨 Received message: {msg.value().decode('utf-8')} from partition {msg.partition()}")

except KeyboardInterrupt:  
    print("🛑 Stopped by user.")

finally:  
    consumer.close()