不能将Kafka用于python

Question

Python:2.6.6

Kafka-蟒蛇:1.4.3

从Kafka.Consumer进口KafkaConsumer

文件“/usr/lib/python2.6/site-packages/kafka/consumer/init.py”，第5行

从kafka.consumer.group导入KafkaConsumer

文件“/usr/lib/python2.6/site-packages/kafka/record/init.py”，第1行

从kafka.record.memory_records导入

文件“/usr/lib/python2.6/site-packages/kafka/record/memory_records.py”，第27行

类DefaultRecordBatchBuilder（DefaultRecordBase,ABCRecordBatchBuilder）：

DefaultRecordBatchBuilder中的文件“/usr/lib/python2.6/site-packages/kafka/record/default_records.py”，第378行

byte_like=(bytes、bytearray、memoryview)，

from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka.errors import KafkaError
import json

class Kafka_producer():

    def __init__(self, kafkahost,kafkaport, kafkatopic):
        self.kafkatopic = kafkatopic
        service_host = kafkahost+":"+kafkaport
        self.producer = KafkaProducer(bootstrap_servers=service_host)

    def sendjsondata(self, params):
        try:
            # parmas_message = json.dumps(params)
            producer = self.producer
            futur = producer.send(self.kafkatopic, params.encode('utf-8'))
            res = futur.get(timeout=60)
            producer.flush()
            producer.close()
        except KafkaError as e:
            print e

if __name__ == '__main__':
    # test = {
    #     "test":"testtets"
    # }
    # Kafka_producer("http://10.25.245.192","9092","nori-log").sendjsondata(test)
    producer = KafkaProducer(bootstrap_servers='10.25.245.192:9092')
    for _ in range(100):
        producer.send('nori-log', {"test":"test_content"})

我把多伦多的版本换成了2.2.1

Answer 1

memoryview类型在Python 2.7中是新的。Python 2.6.6已经有8年的历史了，而且已经有5年没有得到任何支持或安全更新了。您需要升级您的Python安装。