Merge branch 'save_have_xiafa' into 'master'

Save have xiafa

See merge request !22

moment/views/consum.py

@@ -35,7 +35,7 @@ def kafka_consum(topic_name=None):
         #     consumser_obj.seek(partition=TopicPartition(topic_name,int(partition_id.decode())),offset=int(topic_partition_info[partition_id]))
         while True:
             begin = time.time()
-            msg_dict = consumser_obj.poll(timeout_ms=100, max_records=50)
+            msg_dict = consumser_obj.poll(timeout_ms=100, max_records=30)
             for msg_value in msg_dict.values():
                 for msg in msg_value:
                     card_info = json.loads(str(msg.value, encoding="utf8"))
@@ -70,7 +70,6 @@ def kafka_consum(topic_name=None):
                                         auto_follow_user(card_info, after_day=True)
                                         logging.info("当前卡片ID:%s,auto_follow_user子函数消费处理耗时:%f" % (
                                             card_info['card_id'], time.time() - begin))
-
                                     else:
                                         pass
 
@@ -155,14 +154,13 @@ def kafka_consum(topic_name=None):
                                                         auto_follow_user(card_info)
                                                     else:
                                                         logging.info("这条数据将被paas:%s" % card_info)
+
                             else:  # push_time时间未到 需要等待
                                 logging.info("follow [%s,%s,%s,%s],push_time未到,需要等待" % (
                                     str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key)))
                                 save_data_to_kafka(card_info)
-                                pass
                         else:
                             logging.info("此条数据已经消费过了:%s" % card_info)
-                            pass
             logging.info("消费处理耗时:%f" % (time.time() - begin))
     except:
         consumser_obj.close()