84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
# coding:utf-8
|
|
import time
|
|
from multiprocessing import Process
|
|
|
|
from kafka import TopicPartition
|
|
|
|
from settings import settings
|
|
from v2 import *
|
|
from v2.struct_cache import StructCacheFile, StructCacheRedis
|
|
|
|
|
|
class XProcess(Process):
|
|
|
|
def __init__(self, partition, lock, ipsearch, log, rdb=None, event_attr=None):
|
|
super(XProcess, self).__init__()
|
|
# self.daemon = True
|
|
self.partition = partition
|
|
self.lock = lock
|
|
self.ipsearch = ipsearch
|
|
self.rdb = rdb
|
|
self.log = log
|
|
self.event_attr = event_attr
|
|
|
|
def run(self):
|
|
db_client = CK(**settings.CK_CONFIG)
|
|
sketch = Sketch(db_client, StructCacheRedis(self.rdb))
|
|
handler_event = HandlerEvent(db_client, settings.GAME, self.ipsearch)
|
|
handler_user = HandlerUser(db_client, settings.GAME)
|
|
transmitter = Transmitter(db_client, settings.GAME, sketch, self.log, self.lock, self.event_attr,
|
|
self.partition)
|
|
transmitter.start_ping()
|
|
transmitter.add_source(handler_event, 5000, 60)
|
|
transmitter.add_source(handler_user, 500, 60)
|
|
|
|
last_ts = int(time.time())
|
|
consumer, kafka_client = create_consumer(self.partition)
|
|
|
|
for msg in consumer():
|
|
data = msg.value
|
|
type_ = data['#type']
|
|
del data['#type']
|
|
ts = int(time.time())
|
|
|
|
if 'user' in type_:
|
|
# continue
|
|
obj = getattr(handler_user, type_)
|
|
handler_user.receive_data.append(UserAct(obj, data))
|
|
if len(handler_user.receive_data) >= 1000 or last_ts + 60 < ts:
|
|
last_ts = ts
|
|
handler_user.execute()
|
|
|
|
elif 'track' in type_:
|
|
# continue
|
|
if data['#event_name'] == 'pay':
|
|
self.log.info(f'topid->{msg.topic} | partition->{msg.partition} | offset->{msg.offset} | data-> {data}')
|
|
|
|
obj = getattr(handler_event, type_)
|
|
obj(data)
|
|
elif type_ == settings.STOP_SIGNAL:
|
|
# continue
|
|
# 1 小时内有效
|
|
self.log.info(type_)
|
|
if data.get('#time', 0) + 3600 < int(time.time()):
|
|
continue
|
|
kafka_client.close()
|
|
# 停止消费kafka
|
|
self.log.info(f'进程{self.partition} 等待90秒')
|
|
time.sleep(90)
|
|
self.log.info(f'进程{self.partition} 写入数据')
|
|
transmitter.run()
|
|
self.log.info(f'进程{self.partition} 结束')
|
|
|
|
break
|
|
elif type_ == 'test':
|
|
self.log.info(f'topid->{msg.topic} | partition->{msg.partition} | offset->{msg.offset} | data-> {data}')
|
|
else:
|
|
continue
|
|
|
|
transmitter.run()
|
|
|
|
while True:
|
|
time.sleep(5)
|
|
self.log.info(f'消费分区{self.partition} 已结束。。。')
|