Библиотека Aiokafka не потребляет сообщения асинхронно - PullRequest
Новая
       23

Библиотека Aiokafka не потребляет сообщения асинхронно

0 голосов
/ 03 апреля 2019

Я пытаюсь реализовать асинхронную библиотеку Python aiokafka и по какой-то причине не могу обрабатывать сообщения асинхронно

создан асинхронный потребитель, производитель и использует библиотеку 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 Ответ

2 голосов
/ 03 апреля 2019

Сам по себе async for не обрабатывает последовательность параллельно - он просто позволяет сопрограмме приостанавливать работу в ожидании следующего итерируемого асинхронного элемента. Вы можете думать об этом как о серии await с помощью специального метода __anext__, аналогичного обычному для последовательности вызовов __next__.

Но достаточно легко создавать задачи, которые обрабатывают сообщения по мере их поступления. Например: 

async def process(msg):
    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)

async def consume():
    consumer = AIOKafkaConsumer(
        "test_topic", loop=loop, bootstrap_servers='localhost:9092')
    await consumer.start()
    tasks = []
    try:
        async for msg in consumer:
            tasks.append(asyncio.create_task(process(msg))
    finally:
        await consumer.stop()
    await asyncio.gather(*tasks)
...