Skip to content

V
vest
Commit bbaf4bdc authored by 李小芳's avatar 李小芳
Browse files
Options

Merge branch 'save_offset_to_redis' into 'master' 

判断offset是否已经消费了

See merge request !11
parents 398bff5d 0c5a4114
Hide whitespace changes
Inline Side-by-side
Showing with 125 additions and 93 deletions
moment/views/consum.py
View file @ bbaf4bdc
......@@ -19,6 +19,7 @@ from comment.views.comment_fun import comment
from follow.views.follow_fun import follow
from libs.cache import redis_client
from kafka.structs import TopicPartition
from moment.views.process_time import judge_offset_partition_have_consum
def kafka_consum(topic_name=None):
......@@ -40,112 +41,120 @@ def kafka_consum(topic_name=None):
for msg in consume_msg:
card_info = json.loads(str(msg.value, encoding="utf8"))
if card_info['card_type'] == "auto_vest":
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)
# 代表当天数据
current_push_time = card_info['current_push_time']
create_time = card_info['create_time']
nowtime = datetime.datetime.now()
push_time_date = datetime.datetime.strptime(current_push_time, '%Y-%m-%d %H:%M:%S')
if push_time_date <= nowtime: # push_time已经到时间了 需要去下发
# 判断如果当前的push_time 和当前的创建时间一样 需要给push_time下发真的push时间
if current_push_time == create_time:
if card_info['action_type'] == "comment":
auto_comment_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_comment_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
elif card_info['action_type'] == "click":
auto_click_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_click_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
###在这里去判断一下当前的partition和offset是否已经消费过了 如果已经消费了需要直接去掉数据
bol_consum = judge_offset_partition_have_consum(offset=msg.offset, partition=msg.partition)
elif card_info['action_type'] == "follow":
auto_follow_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_follow_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
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)
# 代表当天数据
current_push_time = card_info['current_push_time']
create_time = card_info['create_time']
nowtime = datetime.datetime.now()
push_time_date = datetime.datetime.strptime(current_push_time, '%Y-%m-%d %H:%M:%S')
if push_time_date <= nowtime: # push_time已经到时间了 需要去下发
# 判断如果当前的push_time 和当前的创建时间一样 需要给push_time下发真的push时间
if current_push_time == create_time:
if card_info['action_type'] == "comment":
auto_comment_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_comment_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
elif card_info['action_type'] == "click":
auto_click_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_click_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
elif card_info['action_type'] == "follow":
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
else:
pass
# 当前已下发完 需要判断最新一次的下发时间是否是今天 是的话直接保存起来,l=轮询等待明天再下发
if card_info['have_pust_num'] == card_info['need_pust_num']:
if nowtime.day - push_time_date.day == 0: # 今日的已经下发完了，需要去取之后的
action_type = card_info['action_type']
logging.info("今天已经下发完了[%s,%s,%s,%s]" % (
str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key)))
if action_type == "comment":
auto_comment_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_comment_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
elif action_type == "click":
auto_click_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_comment_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
elif action_type == "follow":
auto_follow_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_comment_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
else:
pass
else:
# 当前已下发完 需要判断最新一次的下发时间是否是今天 是的话直接保存起来,l=轮询等待明天再下发
if card_info['have_pust_num'] == card_info['need_pust_num']:
if nowtime.day - push_time_date.day == 0: # 今日的已经下发完了，需要去取之后的
else: # 还有一种是下发时间已到
action_type = card_info['action_type']
logging.info("今天已经下发完了[%s,%s,%s,%s]" % (
str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key)))
if action_type == "comment":
auto_comment_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_comment_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
elif action_type == "click":
auto_click_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_comment_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
elif action_type == "follow":
auto_follow_user(card_info, after_day=True)
logging.info("当前卡片ID:%s,auto_comment_user子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
else:
pass
else: # 还有一种是下发时间已到
action_type = card_info['action_type']
if card_info['have_pust_num'] < card_info['need_pust_num'] and \
push_time_date < nowtime:
if card_info['have_pust_num'] < card_info['need_pust_num'] and \
push_time_date < nowtime:
if action_type == "comment": # 在这里去调评论的接口
if 'have_comment_number' in card_info and \
card_info['have_comment_number'] < 20:
card_info["have_comment_number"] += 1
logging.info("当前卡片ID:%s,comment1子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
is_success = comment(card_info)
logging.info("comment [%s,%s,%s,%s],当前ID:%s，下发状状态:%s" % (
str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key),
card_info["card_id"], is_success))
logging.info("当前卡片ID:%s,comment2子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
# 调完接口后需要再次去拿新的push_time的时间
auto_comment_user(card_info)
if action_type == "comment": # 在这里去调评论的接口
if 'have_comment_number' in card_info and \
card_info['have_comment_number'] < 20:
card_info["have_comment_number"] += 1
logging.info("当前卡片ID:%s,comment1子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
is_success = comment(card_info)
logging.info("comment [%s,%s,%s,%s],当前ID:%s，下发状状态:%s" % (
str(msg.topic), str(msg.partition), str(msg.offset),
str(msg.key),
card_info["card_id"], is_success))
logging.info("当前卡片ID:%s,comment2子函数消费处理耗时:%f" % (
card_info['card_id'], time.time() - begin))
# 调完接口后需要再次去拿新的push_time的时间
auto_comment_user(card_info)
elif action_type == "click": # 在这里去调点赞的接口
if 'have_click_number' in card_info and card_info['have_click_number'] < 20:
card_info["have_click_number"] += 1
logging.info("当前卡片ID:%s,click1子函数消费处理耗时:%f" % (
card_info["card_id"], time.time() - begin))
is_success = click(card_info)
logging.info("click [%s,%s,%s,%s],当前ID:%s，下发状状态:%s" % (
str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key),
card_info["card_id"], is_success))
logging.info("当前卡片ID:%s,click2子函数消费处理耗时:%f" % (
card_info["card_id"], time.time() - begin))
auto_click_user(card_info)
elif action_type == "click": # 在这里去调点赞的接口
if 'have_click_number' in card_info and card_info[
'have_click_number'] < 20:
card_info["have_click_number"] += 1
logging.info("当前卡片ID:%s,click1子函数消费处理耗时:%f" % (
card_info["card_id"], time.time() - begin))
is_success = click(card_info)
logging.info("click [%s,%s,%s,%s],当前ID:%s，下发状状态:%s" % (
str(msg.topic), str(msg.partition), str(msg.offset),
str(msg.key),
card_info["card_id"], is_success))
logging.info("当前卡片ID:%s,click2子函数消费处理耗时:%f" % (
card_info["card_id"], time.time() - begin))
auto_click_user(card_info)
elif action_type == "follow": # 在这里去调关注的接口
elif action_type == "follow": # 在这里去调关注的接口
if 'have_follow_number' in card_info and \
card_info['have_follow_number'] < 20:
card_info["have_follow_number"] += 1
logging.info("当前卡片ID:%s,follow1子函数消费处理耗时:%f" % (
card_info["card_id"], time.time() - begin))
is_success = follow(card_info)
logging.info("当前卡片ID:%s,follow2子函数消费处理耗时:%f" % (
card_info["card_id"], time.time() - begin))
if 'have_follow_number' in card_info and \
card_info['have_follow_number'] < 20:
card_info["have_follow_number"] += 1
logging.info("当前卡片ID:%s,follow1子函数消费处理耗时:%f" % (
card_info["card_id"], time.time() - begin))
is_success = follow(card_info)
logging.info("当前卡片ID:%s,follow2子函数消费处理耗时:%f" % (
card_info["card_id"], time.time() - begin))
logging.info("follow [%s,%s,%s,%s],当前ID:%s，下发状状态:%s" % (
str(msg.topic), str(msg.partition), str(msg.offset), str(msg.key),
card_info["card_id"], is_success))
logging.info("follow [%s,%s,%s,%s],当前ID:%s，下发状状态:%s" % (
str(msg.topic), str(msg.partition), str(msg.offset),
str(msg.key),
card_info["card_id"], is_success))
auto_follow_user(card_info)
auto_follow_user(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: # 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
logging.info("消费处理耗时:%f" % (time.time() - begin))
except:
consumser_obj.close()
......
moment/views/process_time.py
View file @ bbaf4bdc
......@@ -310,3 +310,26 @@ def get_vest_userid_and_comment(need_comment_num=0, tag_names=[], card_id=0):
except:
logging.error("catch exception,err_msg:%s" % traceback.format_exc())
return []
def judge_offset_partition_have_consum(offset=0, partition=0):
"""
根据当前的offset和分区去判断数据是否已经被消费
:param offset:
:param partition:
:return:
"""
try:
redis_list_data = set()
key = "irrigation_partition_offset_have_consum:" + str(partition)
redis_data = redis_client.get(key)
if redis_data:
redis_list_data = set(json.loads(redis_data))
if offset in redis_list_data:
return False
redis_list_data.add(offset)
redis_client.set(key, json.dumps(list(redis_list_data)))
return True
except:
logging.error("catch exception,err_msg:%s" % traceback.format_exc())
return False
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