2

PyKafka 有以下限制：

传递报告队列是线程本地的：它只会为从当前线程产生的消息提供报告

我正在尝试编写一个脚本，我可以在其中使用一个函数异步发送消息，并通过另一个函数继续接收确认。

以下是功能：

def SendRequest(producer):

        count=0        
        while True:
            count += 1
            producer.produce('test msg', partition_key='{}'.format(count))
            if count == 50000:
                  endtime=datetime.datetime.now()
                  print "EndTime : ",endtime
                  print "Done sending all messages.Waiting for response now"
                  return



def GetResponse(producer):

    count_response=0

    while True:

              try:
                  msg, exc = producer.get_delivery_report(block=False)
                  if exc is not None:
                      count_response+=1
                      print 'Failed to deliver msg {}: {}'.format(
                          msg.partition_key, repr(exc))
                  else:
                      print "Count Res :",count_response
                      count_response+=1

              except Queue.Empty:
                  pass

              except Exception,e:
                  print "Unhandled exception : ",e

线程和多处理没有帮助。以上两个函数需要异步/并行运行。这里应该使用什么方法？

4

1 回答 1

1

问题：我可以在哪里异步发送消息......并继续接收确认

此解决方案asyncio.coroutine将满足您的需求。

注意：有一些缺点！

  • 此asyncio代码至少需要 Python 3.5
  • 对于每条消息，都会创建一个新任务

这实现了class AsyncProduceReport()： 

import asyncio
from pykafka import KafkaClient
import queue, datetime

class AsyncProduceReport(object):
    def __init__(self, topic):
        self.client = KafkaClient(hosts='127.0.0.1:9092')
        self.topic = self.client.topics[bytes(topic, encoding='utf-8')]
        self.producer = self.topic.get_producer(delivery_reports=True)
        self._tasks = 0

    # async
    @asyncio.coroutine
    def produce(self, msg, id):
        print("AsyncProduceReport::produce({})".format(id))
        self._tasks += 1
        self.producer.produce(bytes(msg, encoding='utf-8'))

        # await - resume next awaiting task
        result = yield from self.get_delivery_report(id)

        self._tasks -= 1
        # This return values are passed to self.callback(task)
        return id, result

    def get_delivery_report(self, id):
        """
         This part of a Task, runs as long as of receiving the delivery_report
        :param id: ID of Message
        :return: True on Success else False
        """
        print("{}".format('AsyncProduceReport::get_delivery_report({})'.format(id)))

        while True:
            try:
                msg, exc = self.producer.get_delivery_report(block=False)
                return (not exc, exc)

            except queue.Empty:
                # await - resume next awaiting task
                yield from asyncio.sleep(1)

    @staticmethod
    def callback(task):
        """
         Processing Task Results
        :param task: Holds the Return values from self.produce(...)
        :return: None
        """
        try:
            id, result = task.result()
            print("AsyncProduceReport::callback: Msg:{} delivery_report:{}"
                    .format(id, result))
        except Exception as e:
            print(e)

    def ensure_futures(self):
        """
         This is the first Task
         Creates a new taks for every Message
        :return: None
        """

        # Create 3 Tasks for this testcase
        for id in range(1, 4):
            # Schedule the execution of self.produce(id): wrap it in a future. 
            # Return a Task object.
            # The task will resumed at the next await
            task = asyncio.ensure_future(self.produce('test msg {} {}'
                     .format(id, datetime.datetime.now()), id))

            # Add a Result Callback function
            task.add_done_callback(self.callback)

            # await - resume next awaiting task
            # This sleep value could be 0 - Only for this testcase == 5
            # Raising this value, will give more time for waiting tasks
            yield from asyncio.sleep(5)
            # print('Created task {}...'.format(_id))

        # await - all tasks completed
        while self._tasks > 0:
            yield from asyncio.sleep(1)

用法： 

if __name__ == '__main__':
    client = AsyncProduceReport('topic01')        
    loop = asyncio.get_event_loop()
    loop.run_until_complete(client.ensure_futures())
    loop.close()
    print("{}".format('EXIT main()'))

输入：

AsyncProduceReport::produce(1)
AsyncProduceReport::get_delivery_report(1)
AsyncProduceReport::produce(2)
AsyncProduceReport::get_delivery_report(2)
AsyncProduceReport::callback: Msg:1 delivery_report:(True, None)
AsyncProduceReport::produce(3)
AsyncProduceReport::get_delivery_report(3)
AsyncProduceReport::callback: Msg:2 delivery_report:(True, None)
AsyncProduceReport::callback: Msg:3 delivery_report:(True, None)

用 Python:3.5.3 测试 - pykafka:2.7.0

于 2018-09-20T18:10:57.467 回答