Demo Kết nối Kafka Bizfly Cloud với Producer và Consumer bằ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 \= KAFKA\_CONFIG.copy()  
conf.update({  
    '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()