1.安裝pykafka

pip install pykafka

2.生產者

from pykafka import KafkaClient

from pykafka import KafkaClient

host = ‘192.168.20.203:9092,192.168.20.204:9092,192.168.20.205:9092‘

client = KafkaClient(hosts=host)

print(client.topics)

topic = client.topics["test_kafka_topic"]

for i in range(10):

print(i)

message = "test message test message" + str(i)

message = bytes(message,encoding=‘utf-8‘)

producer = topic.get_producer()

producer.produce(message)

3.消費者

from pykafka import KafkaClient

host = ‘192.168.20.203:9092,192.168.20.204:9092,192.168.20.205:9092‘

client = KafkaClient(hosts=host)

topic=client.topics[‘test_kafka_topic‘]

balanced_consumer = topic.get_balanced_consumer(consumer_group=‘test_kafka_topic‘,auto_commit_enable=True,

zookeeper_connect=‘192.168.20.201:2181,192.168.20.202:2181,192.168.20.203:2181‘)

for messgage in balanced_consumer:

print(messgage)

if messgage is not None:

print(messgage.offset)

print(messgage.value)

4.kafka的命令

創建主題

bin/kafka-topics.sh --create --zookeeper 192.168.183.100:2181 --replication-factor 2 --partitions 3 --topic topicnewtest1

查看主題信息

bin/kafka-topics.sh --describe --zookeeper 192.168.183.100:2181 --topic topicnewtest1

查看kafka中已經創建的主題列表

bin/kafka-topics.sh --list --zookeeper 192.168.183.100:2181

刪除主題

bin/kafka-topics.sh --delete --zookeeper 192.168.183.100:2181 --topic topictest1

查看主題中的消息

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group test_kafka_topic --topic test_kafka_topic --zookeeper 192.168.20.203:2181

增加分區

bin/kafka-topics.sh --alter --zookeeper 192.168.183.100:2181 --topic topicnewtest1 --partitions 5

使用kafka自帶的生產者客戶端腳本

bin/kafka-console-producer.sh --broker-list 192.168.183.102:9092,192.168.183.103:9092 --topic topicnewtest1

使用kafka自帶的消費者客戶端腳本

bin/kafka-console-consumer.sh --zookeeper 192.168.183.100:2181 --from-beginning --topic topicnewtest1

總結

以上是生活随笔為你收集整理的python下载模块命令_python kafka模块操作命令集合的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。