data_agent.py 文件源码

def batch_fetch(self, queue, event, linger_ms, max_queued_messages):
        if queue.qsize() < max_queued_messages:
            event.wait(linger_ms / 1000)
        if event.is_set():
            event.clear()
        batch_msgs = [queue.get() for _ in range(queue.qsize())]
        return batch_msgs

    #@profile