Apache Kafka是一个基于发布-订阅模式的消息传递系统。在Kafka中,消息的状态可以分为以下两个方面:生产者端消息状态和消费者端消息状态。
生产者端消息状态表示了发送到Kafka的消息的状态。我们可以使用回调(callback)函数来检查生产者端消息状态。
下面是一个示例代码:
from kafka import KafkaProducer
def on_send_success(metadata):
print('Message sent to partition %d with offset %d' % (metadata.partition, metadata.offset))
def on_send_error(excp):
print('Error while sending message:', excp)
producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer.send('test-topic', key=b'key', value=b'value').add_callback(on_send_success).add_errback(on_send_error)
当消息成功发送到Kafka时,on_send_success函数将被调用,消息元数据(metadata)会包含分区和偏移量信息。 当发送消息时遇到错误,on_send_error函数会被调用。
消费者端消息状态表示了消费者接收到的消息的状态。 在Kafka中,消费者可以通过控制偏移量(offset)来管理消费进度和状态。
下面是一个示例代码:
from kafka import KafkaConsumer
consumer = KafkaConsumer('test-topic', bootstrap_servers='localhost:9092', group_id='my-group')
for message in consumer:
print(message.topic, message.partition, message.offset, message.key, message.value)
# 提交偏移量
consumer.commit()
在上面的示例中,我们创建了一个Kafka消费者对象,使用组ID“my-group”来表示每个消费者的标识。 然后我们使用for循环来消费消息,并打印出消息的一些属性(如topic,偏移量,键值对等)。 在消息被处理之后,我们提交偏移