auto_vest

moment/views/consum.py

@@ -26,8 +26,8 @@ def kafka_consum(topic_name=None):
     consumser_obj.subscribe([topic_name, ])
     try:
         while True:
+            begin = datetime.datetime.now()
             msg_dict = consumser_obj.poll(timeout_ms=100, max_records=30)
-            consumser_obj.commit_async()
             for msg_key in msg_dict:
                 consume_msg = msg_dict[msg_key]
                 for msg in consume_msg:
@@ -111,6 +111,7 @@ def kafka_consum(topic_name=None):
                             logging.info("push_time时间未到========================需要等待")
                             save_data_to_kafka(card_info)
                             pass
+            print(datetime.datetime.now() - begin)
 
     except:
         consumser_obj.close()

moment/views/process_time.py

@@ -244,8 +244,6 @@ def save_data_to_kafka(card_info):
     try:
         topic = settings.KAFKA_TOPIC_NAME
         producer.send(topic, json.dumps(card_info).encode())
-        producer.close()
-
     except:
         logging.error("catch exception,err_msg:%s" % traceback.format_exc())