kafkaReceiveTest.py 1011 B

123456789101112
  1. from kafka import KafkaConsumer
  2. # tag_sink_topic\tag_atom_topic
  3. consumer = KafkaConsumer('tag_flag_topic',
  4. group_id='test115', # 一个组消费一次
  5. auto_offset_reset='latest', # 从最新数据读取,earliest,latest
  6. bootstrap_servers=['kafka-0.kafka-headless.rxdpdev.svc.k5.bigtree.zone:9092', 'kafka-1.kafka-headless.rxdpdev.svc.k5.bigtree.zone:9092', 'kafka-2.kafka-headless.rxdpdev.svc.k5.bigtree.zone:9092']
  7. # bootstrap_servers=['kafka-0.kafka-headless.aimpdev.svc.k5.bigtree.zone:9092', 'kafka-1.kafka-headless.aimpdev.svc.k5.bigtree.zone:9092', 'kafka-2.kafka-headless.aimpdev.svc.k5.bigtree.zone:9092']
  8. # bootstrap_servers=['kafka-0.kafka-headless.aimptest.svc.k5.bigtree.zone:9092', 'kafka-1.kafka-headless.aimptest.svc.k5.bigtree.zone:9092', 'kafka-2.kafka-headless.aimptest.svc.k5.bigtree.zone:9092']
  9. )
  10. for msg in consumer:
  11. print(msg.value)