我一直在使用AIOKafka一段时间，直到今天我都没有遇到任何问题。
当我尝试使用TypeError发送消息时，出现奇怪的AIOKafkaProducer.send_and_wait。我还将此问题作为问题发布在AIOKafka的github存储库上，但看起来它们是非活动的。也许有人可以帮助我。
这是代码：
import asyncio
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
loop = asyncio.get_event_loop()
producer = AIOKafkaProducer(loop=loop, bootstrap_servers="localhost:9092")
async def _initialize(prod, future):
await prod.start()
await prod.send_and_wait("main_topic", str.encode("hello!!"))
future = asyncio.Future()
task = asyncio.ensure_future(_initialize(producer, future))
loop.run_until_complete(task)
print("loop ended!")
loop.close()
这是我收到的错误消息：
yilmazali@yilmazali:~$ python3 aiokafkatest.py
Unexpected error in sender routine
Traceback (most recent call last):
File "/usr/local/lib/python3.6/dist-packages/aiokafka/producer/producer.py", line 374, in _sender_routine
task.result()
File "/usr/local/lib/python3.6/dist-packages/aiokafka/producer/producer.py", line 418, in _send_produce_req
response = yield from self.client.send(node_id, request)
File "/usr/local/lib/python3.6/dist-packages/aiokafka/client.py", line 415, in send
request, expect_response=expect_response)
File "/usr/local/lib/python3.6/dist-packages/aiokafka/conn.py", line 165, in send
message = header.encode() + request.encode()
File "/usr/local/lib/python3.6/dist-packages/kafka/util.py", line 159, in __call__
return self.method()(self.target(), *args, **kwargs)
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/struct.py", line 42, in _encode_self
[self.__dict__[name] for name in self.SCHEMA.names]
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 132, in encode
for i, field in enumerate(self.fields)
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 132, in <listcomp>
for i, field in enumerate(self.fields)
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 170, in encode
[self.array_of.encode(item) for item in items]
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 170, in <listcomp>
[self.array_of.encode(item) for item in items]
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 132, in encode
for i, field in enumerate(self.fields)
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 132, in <listcomp>
for i, field in enumerate(self.fields)
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 170, in encode
[self.array_of.encode(item) for item in items]
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 170, in <listcomp>
[self.array_of.encode(item) for item in items]
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 132, in encode
for i, field in enumerate(self.fields)
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 132, in <listcomp>
for i, field in enumerate(self.fields)
File "/usr/local/lib/python3.6/dist-packages/kafka/protocol/types.py", line 93, in encode
return Int32.encode(len(value)) + value
TypeError: object of type '_io.BytesIO' has no len()
我没有更改我的kafka结构或库。我的卡夫卡经纪人看起来不错。我可以使用shell脚本生成/使用消息。
在过去2-3个月中，我在AIOKafka上没有遇到任何问题，上面的代码运行良好。突然出现此错误，我想知道问题出在哪里。
任何帮助将不胜感激。
给予最亲切的问候，
阿里
-
更新：我们在朋友的计算机上运行了此代码段，它工作正常。我向外面做广告我的卡夫卡，她用上面的代码成功地写了我当地的卡夫卡主题。两台机器上的AIOKafka库版本均为0.4.0。两台计算机上的asyncio版本均为3.4.3。简而言之，问题不在于我的kafka或库。我的机器出了点问题，但是上帝知道是什么原因引起的。
最佳答案
终于在我的机器上工作了。我只是卸载并安装了aiokafka模块。
尽管我对这种解决方案不满意，并且想更深入地探究问题的核心，但我很高兴现在可以继续进行工作。
希望这对遇到同样问题的陌生人有所帮助。