68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
|
import traceback
|
|||
|
|
from kafka import KafkaProducer, KafkaConsumer,TopicPartition
|
|||
|
|
from kafka.errors import kafka_errors
|
|||
|
|
import json
|
|||
|
|
def get_left_cnt(consumer,topic):
|
|||
|
|
partitions = [TopicPartition(topic, p) for p in consumer.partitions_for_topic(topic)]
|
|||
|
|
|
|||
|
|
# total
|
|||
|
|
toff = consumer.end_offsets(partitions)
|
|||
|
|
toff = [(key.partition, toff[key]) for key in toff.keys()]
|
|||
|
|
toff.sort()
|
|||
|
|
|
|||
|
|
# current
|
|||
|
|
coff = [(x.partition, consumer.committed(x)) for x in partitions]
|
|||
|
|
coff.sort()
|
|||
|
|
|
|||
|
|
# cal sum and left
|
|||
|
|
toff_sum = sum([x[1] for x in toff])
|
|||
|
|
cur_sum = sum([x[1] for x in coff if x[1] is not None])
|
|||
|
|
left_sum = toff_sum - cur_sum
|
|||
|
|
|
|||
|
|
return left_sum
|
|||
|
|
def getAllRecords(consumer,topics):
|
|||
|
|
leftCnt = 0
|
|||
|
|
for topic in topics[0:2]:
|
|||
|
|
leftCnt+=get_left_cnt(consumer,topic)
|
|||
|
|
out = []
|
|||
|
|
if leftCnt == 0:
|
|||
|
|
return []
|
|||
|
|
for ii,msg in enumerate(consumer):
|
|||
|
|
consumer.commit()
|
|||
|
|
out.append(msg)
|
|||
|
|
if ii== (leftCnt-1):
|
|||
|
|
break###断流或者到终点
|
|||
|
|
return out
|
|||
|
|
def detector(par):
|
|||
|
|
consumer = KafkaConsumer(
|
|||
|
|
bootstrap_servers=par['server'],
|
|||
|
|
group_id=par['group_id'],
|
|||
|
|
#auto_offset_reset='earliest',
|
|||
|
|
auto_offset_reset='latest',
|
|||
|
|
#isolation_level = 'read_committed',
|
|||
|
|
#enable_auto_commit=True
|
|||
|
|
)
|
|||
|
|
consumer.subscribe( par['topic'][0:2])
|
|||
|
|
print( ' Start kafka ')
|
|||
|
|
msgs = getAllRecords(consumer,par['topic'])
|
|||
|
|
print( 'getover cnt',len(msgs))
|
|||
|
|
for ii,msg in enumerate(msgs):
|
|||
|
|
print(msg)
|
|||
|
|
try:
|
|||
|
|
print('##'*10,ii)
|
|||
|
|
taskInfos = eval(msg.value.decode('utf-8'))
|
|||
|
|
print(taskInfos )
|
|||
|
|
except:
|
|||
|
|
print('**'*10,'wrong',ii)
|
|||
|
|
print(msg.value.decode('utf-8'))
|
|||
|
|
if __name__ == '__main__':
|
|||
|
|
par={};
|
|||
|
|
###topic0--在线，topic1--离线
|
|||
|
|
|
|||
|
|
#par['server']='212.129.223.66:9092';par['topic']=('thsw','thsw2','testReturn');par['group_id']='test';
|
|||
|
|
|
|||
|
|
par['server']='101.132.127.1:19092';par['topic']=('alg-online-tasks', 'alg-task-results','alg-offline-tasks');par['group_id']='testww';
|
|||
|
|
#par['server']='101.132.127.1:19092';par['topic']=('alg-online-tasks','alg-task-results','alg-offline-tasks');par['group_id']='testW11';
|
|||
|
|
par['kafka']='mintors/kafka'
|
|||
|
|
detector(par)
|