from pykafka import KafkaClientclass KafkaProduct():def __init__(self,hosts,topic):"""初始化实例:param hosts: 连接地址:param topic:"""self.__client = KafkaClient(hosts=hosts)self.__topic = self.__client.topics[topic.encode()]def __set_topic(self, topic):self.__topic = self.__client.topics[topic.encode()]def set_topic(self, topic):"""设置topic:param topic::return:"""self.__set_topic(topic)def get_topics(self):"""获取当前所有topic:return:"""return self.__client.topicsdef get_topic(self):"""获取当前topic:return:"""return self.__topicdef Producer(self):"""生产者对象:return:"""with self.__topic.get_producer(delivery_reports=True) as producer:next_data = ''while True:if next_data:producer.produce(str(next_data).encode())next_data = yield Truedef send_data(self,data):"""发送数据:param datas:需要传入的可迭代对象:return:"""c = self.Producer()next(c)# for i in datas:c.send(data)if __name__ == '__main__':hosts = "localhost:9092" #连接hoststopic = "test_523"K = KafkaProduct(hosts=hosts, topic=topic)  ##K.set_topic("test")  #切换设置新的topicK.get_topic()  #获取当前设置的topic#K.get_topics() #获取所有topicdata = 100912 #要发送的可迭代对象K.send_data(data)