PythonからKafkaに接続する方法

Kafkaへ接続するには、Kafka-Pythonライブラリを使用します。以下に、Kafkaに接続してメッセージを送信する方法を示す簡単なコードサンプルを示します。

from kafka import KafkaProducer, KafkaConsumer
# 连接Kafka生产者
producer = KafkaProducer(bootstrap_servers='localhost:9092')
# 发送消息
producer.send('my_topic', b'Hello, Kafka!')
# 关闭生产者连接
producer.close()
# 连接Kafka消费者
consumer = KafkaConsumer('my_topic', bootstrap_servers='localhost:9092')
# 消费消息
for message in consumer:
print(message.value.decode('utf-8'))
# 关闭消费者连接
consumer.close()

上例では最初にKafkaProducerクラスを使用し、Kafkaに接続し、send()メソッドでmy_topicというトピックにメッセージを送っています。

そして、KafkaConsumerクラスを利用してKafkaに接続し、consumer()メソッドを利用してmy_topicのトピックを購読します。そして、コンシューマオブジェクトをforループで回してメッセージを受け取り、valueプロパティを利用してメッセージの値を取得します。

最後に、生産者と消費者の接続を閉じます。

ただし、上記の例では、Kafkaサーバーがローカルで実行されており、ポート番号が9092であることを前提としています。Kafkaサーバーの設定に応じて、適切に変更してください。