我正在尝试实现python-aiokafka异步库,由于某些原因,我无法异步处理消息。
我创建了async consumer、producer并使用asyncio python库。
环境:
python 3.7.2
aiokafka==0.5.1
kafka-python==1.4.3
消费者:
from aiokafka import AIOKafkaConsumer
import asyncio
import json
import ast
loop = asyncio.get_event_loop()
async def consume():
consumer = AIOKafkaConsumer(
"test_topic", loop=loop, bootstrap_servers='localhost:9092')
# Get cluster layout and topic/partition allocation
await consumer.start()
try:
async for msg in consumer:
sleep_time = ast.literal_eval(json.loads(msg.value))
print('before sleep %s' % sleep_time)
await asyncio.sleep(sleep_time)
print('after sleep %s' % sleep_time)
finally:
await consumer.stop()
loop.run_until_complete(consume())
制作人:
import json
import uuid
from kafka import KafkaProducer, KafkaConsumer
class KafkaClient(object):
def __init__(self, topic_name=None, consume=True):
"""
Initial consumer and producer for Kafka
:param topic_name: consumer topic name
"""
self.topic_name = topic_name
if topic_name is not None:
self.kafka_connect(topic_name, source='SOURCE')
self.producer = KafkaProducer(bootstrap_servers='localhost:9092',
key_serializer=str.encode,
value_serializer=lambda m: json.dumps(m).encode('utf-8'))
def publish_message(self, topic_name, message, extra_data=None):
try:
msg_uid = str(uuid.uuid1())
self.producer.send(topic_name, value=json.dumps(message))
self.producer.flush()
print('Message published [msg_uid]: %s' % msg_uid)
return True
except Exception as err:
print(err)
return False
k = KafkaClient()
for i in range(0, 1):
k.publish_message('test_topic', 5)
k.publish_message('test_topic', 3)
k.publish_message('test_topic', 1)
预期结果:
该过程将打印:
before sleep 5
before sleep 3
before sleep 1
after sleep 1
after sleep 3
after sleep 5
实际结果:
过程打印
before sleep 5
after sleep 5
before sleep 3
after sleep 3
before sleep 1
after sleep 1
1条答案
按热度按时间h7wcgrx31#
独自一人,
async for
不并行处理序列—它只允许协同程序在等待async iterable生成下一项时挂起。你可以把它看作是一系列await
他在街上__anext__
特殊方法,类似于普通方法,是对__next__
.但是在消息到达时生成处理消息的任务非常容易。例如: