消息队列分级并设置动态消费者

BaseDataMaintenance/maintenance/dataflow_mq.py

@@ -189,7 +189,7 @@ class Dataflow_ActivteMQ_attachment(Dataflow_attachment):
             listener_p = Process(target=self.start_attachment_listener,args=(5,self.mq_attachment_his))
             listener_p.start()
 
-        listener_p = Process(target=dynamic_listener,args=(["dataflow_attachment","dataflow_attachment_fix","dataflow_attachment_his"],6,self.AttachmentMQListener,lambda comsumer_count,queue_name:self.start_attachment_listener(comsumer_count,queue_name),5))
+        listener_p = Process(target=dynamic_listener,args=(["dataflow_attachment","dataflow_attachment_fix","dataflow_attachment_his"],6,self.AttachmentMQListener,lambda _dict:self.attachment_listener_handler(_dict),5))
         listener_p.start()
 
         # listener_p = Process(target=self.start_attachment_listener)
@@ -969,7 +969,7 @@ class Dataflow_ActivteMQ_extract(Dataflow_extract):
             for ii in range(1):
                 listener_p = Process(target=self.start_extract_listener,args=(5,self.mq_extract_his))
                 listener_p.start()
-            listener_p = Process(target=dynamic_listener,args=(["dataflow_extract","dataflow_extract_fix","dataflow_extract_his"],8,self.ExtractListener,lambda comsumer_count,queue_name:self.start_extract_listener(comsumer_count,queue_name),5))
+            listener_p = Process(target=dynamic_listener,args=(["dataflow_extract","dataflow_extract_fix","dataflow_extract_his"],8,self.ExtractListener,lambda _dict,result_queue:self.comsumer_handle(_dict,result_queue),5))
             listener_p.start()
             listener_p_ai = Thread(target=self.start_extract_AI_listener)
             listener_p_ai.start()