Skip to content

V
vest
Commit 9a0d441a authored by lixiaofang's avatar lixiaofang
Browse files
Options

提升poll的数据

parents 5aa7237f fa2739c8
Show whitespace changes
Inline Side-by-side
Showing with 47 additions and 16 deletions
moment/views/consum.py
View file @ 9a0d441a
......@@ -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):
......@@ -35,11 +36,14 @@ def kafka_consum(topic_name=None):
while True:
begin = time.time()
msg_dict = consumser_obj.poll(timeout_ms=100, max_records=50)
for msg_key in msg_dict:
consume_msg = msg_dict[msg_key]
for msg in consume_msg:
for msg_value in msg_dict.values():
for msg in msg_value:
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(offset=msg.offset, partition=msg.partition)
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)
......@@ -104,7 +108,8 @@ def kafka_consum(topic_name=None):
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),
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))
......@@ -112,13 +117,15 @@ def kafka_consum(topic_name=None):
auto_comment_user(card_info)
elif action_type == "click": # 在这里去调点赞的接口
if 'have_click_number' in card_info and card_info['have_click_number'] < 20:
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),
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))
......@@ -136,7 +143,8 @@ def kafka_consum(topic_name=None):
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),
str(msg.topic), str(msg.partition), str(msg.offset),
str(msg.key),
card_info["card_id"], is_success))
auto_follow_user(card_info)
......
moment/views/process_time.py
View file @ 9a0d441a
......@@ -60,20 +60,20 @@ def get_one_six_days_random_time(frmt='%Y-%m-%d %H:%M:%S', num_days=0, action_ty
action_num = random.randint(5, 10)
if num_days <= 15 and num_days > 1 and action_type in ("follow", "click") and int(content_level) < 3:
action_num = random.randint(0, 1)
action_num = random.randint(1, 2)
if num_days <= 15 and num_days > 1 and action_type in ("follow") and int(content_level) >= 3:
action_num = random.randint(0, 5)
action_num = random.randint(1, 5)
if num_days == 1 and action_type in ("click") and int(content_level) >= 3:
action_num = random.randint(6, 12)
if num_days <= 15 and num_days > 1 and action_type in ("click") and int(content_level) >= 3:
action_num = random.randint(0, 6)
action_num = random.randint(1, 6)
if num_days >= 1 and num_days <= 6 and action_type in ("comment"):
if int(content_level) <= 3:
action_num = random.randint(0, 1)
action_num = random.randint(1, 2)
else:
action_num = random.randint(2, 4)
......@@ -104,24 +104,24 @@ def get_ten_last_days_random_time(num_days=None, frmt='%Y-%m-%d %H:%M:%S', conte
lastday = datetime.datetime(now.year, now.month, now.day, 23, 0, 0)
add_number = 0
if num_days > 15 and action_type in ("follow"):
action_num = random.randint(0, 2)
action_num = random.randint(1, 2)
add_number = 10
elif num_days > 15 and action_type in ("click"):
if content_level < 3:
action_num = random.randint(0, 1)
action_num = random.randint(1, 2)
add_number = 6
else:
action_num = random.randint(0, 2)
action_num = random.randint(1, 2)
add_number = 5
elif num_days > 6 and action_type in ("comment"):
if content_level <= 3:
action_num = random.randint(0, 1)
action_num = 1
add_number = 10
else:
action_num = random.randint(0, 2)
action_num = random.randint(1, 2)
add_number = 10
else:
......@@ -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