python使用kafka

以你之姓@ 2023-07-13 13:53 3阅读 0赞

python使用kafka

1.安装包：

pip install kafka
    或
    pip install python-kafka

文档：  
`https://kafka-python.readthedocs.io/en/master/usage.html`

2.生产者：

#coding:utf-8
    
    from kafka import KafkaProducer
    
    producer = KafkaProducer(bootstrap_servers=['192.168.137.51:9092']) #此处ip可以是多个['0.0.0.1:9092','0.0.0.2:9092','0.0.0.3:9092' ]
    
    for i in range(3):
        msg = "msg%d" % i
        producer.send('mtest', msg)
    
    producer.close()

3.消费者

from kafka import KafkaConsumer
    
    consumer = KafkaConsumer('mtest', bootstrap_servers=['192.168.137.51:9092'])
    for message in consumer:
        print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
        message.offset, message.key,
        message.value))

定时拉取消息， poll

from kafka import KafkaConsumer
    import time
    
    consumer = KafkaConsumer(bootstrap_servers=['192.168.137.51:9092',])
    consumer.subscribe(topics=['mtest'])
    index = 0
    while True:
        msg = consumer.poll(timeout_ms=5)
        print msg
        time.sleep(2)
        index += 1
        print '--------poll index is %s----------' % index

# To consume latest messages and auto-commit offsets
    consumer = KafkaConsumer('my-topic',
                             group_id='my-group',
                             bootstrap_servers=['localhost:9092'])
    for message in consumer:
        # message value and key are raw bytes -- decode if necessary!
        # e.g., for unicode: `message.value.decode('utf-8')`
        print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                                              message.offset, message.key,
                                              message.value))
    
    # consume earliest available messages, don't commit offsets
    KafkaConsumer(auto_offset_reset='earliest', enable_auto_commit=False)
    
    # consume json messages
    KafkaConsumer(value_deserializer=lambda m: json.loads(m.decode('ascii')))
    
    # consume msgpack
    KafkaConsumer(value_deserializer=msgpack.unpackb)
    
    # StopIteration if no message after 1sec
    KafkaConsumer(consumer_timeout_ms=1000)
    
    # Subscribe to a regex topic pattern
    consumer = KafkaConsumer()
    consumer.subscribe(pattern='^awesome.*')
    
    # Use multiple consumers in parallel w/ 0.9 kafka brokers
    # typically you would run each on a different server / process / CPU
    consumer1 = KafkaConsumer('my-topic',
                              group_id='my-group',
                              bootstrap_servers='my.server.com')
    consumer2 = KafkaConsumer('my-topic',
                              group_id='my-group',
                              bootstrap_servers='my.server.com')