app.py

@@ -2,8 +2,6 @@
 import time
 from multiprocessing import Process
 
-from kafka import TopicPartition
-
 from settings import settings
 from v2 import *
 from v2.struct_cache import StructCacheFile, StructCacheRedis
@@ -35,16 +33,16 @@ class XProcess(Process):
         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']
+        for topic, msg in consumer():
+            # print(msg)
+            type_ = msg['#type']
+            del msg['#type']
             ts = int(time.time())
 
             if 'user' in type_:
                 # continue
                 obj = getattr(handler_user, type_)
-                handler_user.receive_data.append(UserAct(obj, data))
+                handler_user.receive_data.append(UserAct(obj, msg))
                 if len(handler_user.receive_data) >= 1000 or last_ts + 60 < ts:
                     last_ts = ts
                     handler_user.execute()
@@ -52,12 +50,12 @@ class XProcess(Process):
             elif 'track' in type_:
                 # continue
                 obj = getattr(handler_event, type_)
-                obj(data)
+                obj(msg)
             elif type_ == settings.STOP_SIGNAL:
                 # continue
                 # 1 小时内有效
                 self.log.info(type_)
-                if data.get('#time', 0) + 3600 < int(time.time()):
+                if msg.get('#time', 0) + 3600 < int(time.time()):
                     continue
                 kafka_client.close()
                 # 停止消费kafka
@@ -68,8 +66,6 @@ class XProcess(Process):
                 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
 

single_process.py

@@ -1,8 +1,6 @@
-# coding:utf-8
 import time
 
 import redis
-from kafka import TopicPartition
 
 from settings import settings
 from v2 import *
@@ -18,7 +16,6 @@ from v2.log import logger
 rdb = redis.Redis(**settings.REDIS_CONF)
 event_attr = EventAttr(rdb)
 
-partition = -1
 
 def run():
     db_client = CK(**settings.CK_CONFIG)
@@ -26,21 +23,25 @@ def run():
     handler_event = HandlerEvent(db_client, settings.GAME, ipsearch)
     handler_user = HandlerUser(db_client, settings.GAME)
     transmitter = Transmitter(db_client, settings.GAME, sketch, logger, lock, event_attr)
-    transmitter.add_source(handler_event, 1000, 10)
+    transmitter.add_source(handler_event, 10000, 10)
     transmitter.add_source(handler_user, 1000, 10)
     last_ts = int(time.time())
-    consumer, kafka_client = create_consumer(partition)
+    consumer, kafka_client = create_consumer(-1)
 
-    for msg in consumer():
-        data = msg.value
-        type_ = data['#type']
-        del data['#type']
+    for topic, msg in consumer():
+        # print(msg)
+        type_ = msg['#type']
+        # if msg['#event_name'] != 'pay':
+        #     continue
+        # print(msg)
+
+        del msg['#type']
         ts = int(time.time())
 
         if 'user' in type_:
             # continue
             obj = getattr(handler_user, type_)
-            handler_user.receive_data.append(UserAct(obj, data))
+            handler_user.receive_data.append(UserAct(obj, msg))
             if len(handler_user.receive_data) >= 1000 or last_ts + 60 < ts:
                 last_ts = ts
                 handler_user.execute()
@@ -48,33 +49,12 @@ def run():
         elif 'track' in type_:
             # continue
             obj = getattr(handler_event, type_)
-            obj(data)
-        elif type_ == settings.STOP_SIGNAL:
-            # continue
-            # 1 小时内有效
-            print(type_)
-            if data.get('#time', 0) + 3600 < int(time.time()):
-                continue
-            kafka_client.close()
-            # 停止消费kafka
-            print(f'进程{msg.partition} 等待90秒')
-            time.sleep(1)
-            print(f'进程{msg.partition} 写入数据')
-            transmitter.run()
-            print(f'进程{msg.partition} 结束')
-
-            break
-        elif type_ == 'test':
-            print(f'topid->{msg.topic} | partition->{msg.partition} | offset->{msg.offset} | data-> {data}')
+            obj(msg)
         else:
             continue
 
         transmitter.run()
 
-    while True:
-        time.sleep(5)
-        print(f'消费分区{partition} 已结束。。。')
-
 
 if __name__ == '__main__':
     run()

v2/consumer.py

@@ -16,9 +16,8 @@ def create_consumer(partition=-1):
             c.subscribe([settings.SUBSCRIBE_TOPIC])
         for msg in c:
             # print(msg)
-            yield msg
-            # topic = msg.topic
-            # val = msg.value
-            # yield topic, val
+            topic = msg.topic
+            val = msg.value
+            yield topic, val
 
     return consumer, c