判断等级

moment/views/consum.py

@@ -28,7 +28,7 @@ def kafka_consum(topic_name=None):
                                   auto_commit_interval_ms=1, group_id="vest")
     consumser_obj.subscribe([topic_name, ])
     try:
-        redis_topic_partition_name = "vest:topic_name:" + str(topic_name)
+        # redis_topic_partition_name = "vest:topic_name:" + str(topic_name)
         # topic_partition_info = redis_client.hgetall(redis_topic_partition_name)
         # for partition_id in topic_partition_info:
         #     print (int(partition_id.decode()),topic_name)
@@ -46,7 +46,7 @@ def kafka_consum(topic_name=None):
                         if bol_consum:
                             logging.info("消费到新数据了[%s,%s,%s,%s],get card_info:%s" % (
                                 str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key), card_info))
-                            redis_client.hset(redis_topic_partition_name, msg.partition, msg.offset)
+                            # redis_client.hset(redis_topic_partition_name, msg.partition, msg.offset)
                             # 代表当天数据
                             current_push_time = card_info['current_push_time']
                             create_time = card_info['create_time']
@@ -154,6 +154,9 @@ def kafka_consum(topic_name=None):
                                     str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key)))
                                 save_data_to_kafka(card_info)
                                 pass
+                        else:
+                            logging.info("此条数据已经消费过了")
+                            pass
             logging.info("消费处理耗时:%f" % (time.time() - begin))
     except:
         consumser_obj.close()