def enqueue(self, queue_event_list, max_queued_messages):
if len(queue_event_list) == 0:
return
while True:
try:
# get msg
task_msg = _reportQueue.get()
if not task_msg:
continue
dataid, org, ip = task_msg[0][-3:]
logger.debug('recv msg, org: %s dataid: %s' %(org, dataid))
# enqueue
for (q, flush_ready_event) in queue_event_list:
if not q.full():
q.put_nowait(task_msg)
else:
logger.error("queue full")
if q.qsize() >= max_queued_messages and not flush_ready_event.is_set():
flush_ready_event.set()
except Exception, e:
logger.error(e)
评论列表
文章目录