leguadmin/pointapi
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

update

Browse Source
This commit is contained in:
wu hao 2021-04-02 01:30:28 +08:00
parent 8dcf85dee9
commit 6e8e553592
1 changed files with 3 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
output/kafka_p.py
View File

@ -13,22 +13,13 @@ class ToKafka(BaseOutput):
""" """
def __init__(self, conf): def __init__(self, conf):
self.__topic_name = None self.topic_name = None
self.__producer = KafkaProducer(**conf) self.__producer = KafkaProducer(**conf)
self.partition = 15 self.__partition = 15
@property
def topic_name(self):
return self.__topic_name
@topic_name.setter
def topic_name(self, topic_name):
self.__topic_name = topic_name
# self.__producer.partitions_for(topic_name)
def send(self, msg): def send(self, msg):
# msg="""{"#type": "user_set", "#ip": "192.168.1.1", "#time": "2021-04-01 17:54:28.084", "properties": {"herostate": [["fffgsa", 2, 3, 4, 5], ["ddd", 4, 5, 6, 8]], "lv": 60, "#user_id": "fvdsvgdsf"}, "#distinct_id": "ABCDEF123456", "#account_id": "TA10001"}""" # msg="""{"#type": "user_set", "#ip": "192.168.1.1", "#time": "2021-04-01 17:54:28.084", "properties": {"herostate": [["fffgsa", 2, 3, 4, 5], ["ddd", 4, 5, 6, 8]], "lv": 60, "#user_id": "fvdsvgdsf"}, "#distinct_id": "ABCDEF123456", "#account_id": "TA10001"}"""
try: try:
self.__producer.send(self.__topic_name, msg, partition=random.randint(0, self.partition)) self.__producer.send(self.topic_name, msg, partition=random.randint(0, self.__partition))
except Exception as e: except Exception as e:
print(e) print(e)