python之Pykafka之异步发送消息和接收确认

PyKafka 的限制是:

delivery report queue is thread-local: it will only serve reports for messages which were produced from the current thread

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

函数如下:

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 

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

请您参考如下方法:

Question: where I can asynchronously send messages ... and keep receiving acknowledgments

这个带有 asyncio.coroutine 的解决方案将满足您的需求。

Note: There are a few drawbacks!

• This asyncio code needs at least Python 3.5
• For every Message, a new Task is created

这实现了 类 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) 

Usage:

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

Qutput:

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 测试