Commit 85cc0ccc authored by 李小芳's avatar 李小芳

Merge branch 'change_time' into 'master'

Change time

See merge request !30
parents eb1232e0 f97b2184
......@@ -47,11 +47,11 @@ def kafka_consum(topic_name=None):
card_info = json.loads(str(msg.value, encoding="utf8"))
if card_info['card_type'] == "auto_vest":
###在这里去判断一下当前的partition和offset是否已经消费过了 如果已经消费了需要直接去掉数据
# bol_consum = judge_offset_partition_have_consum(card_info=card_info, offset=msg.offset,
# partition=msg.partition)
bol_consum = judge_offset_partition_have_consum(card_info=card_info, offset=msg.offset,
partition=msg.partition)
##在这里先判断当前的数据是否已经存在,存在的话直接PaaS,不存在的话再做下边的处理
# bol_consum = judge_data_have_in_redis(card_info)
bol_consum = True
# bol_consum = True
if bol_consum:
logging.info("消费到新数据了[%s,%s,%s,%s],get card_id:%s,create_time:%s,current_push_time:%s" % (
str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key),
......@@ -140,7 +140,7 @@ def kafka_consum(topic_name=None):
str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key)))
save_data_to_kafka(card_info)
else:
logging.info("此条数据已经被paas了:%s" % card_info)
logging.info("此条数据已经被paas了:%s" % card_info['card_id'])
logging.info("消费处理耗时:%f" % (time.time() - begin))
except:
consumser_obj.close()
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment