kafka相关概念及应用实战(Python代码)

什么是kafka

• Apache Kafka是一个分布式的流处理平台，由Scala写成，并由Apache软件基金会开发的一个开源消息系统项目，该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。
• Kafka是一个分布式消息中间件。

kafka的整体架构和主要概念

kafka架构.png

• Producer， 向队列中发送消息。Produce负责选择将一条记录添加到topic中的某个partition中。

• Consumer，从队列中消费消息。每个Consumer都要有一个Consumer group。

• Broker，每一个kafka实例就是一个broker，一个broker可以有多个topic。

• topic，用于分类消息。topic其本质是一个目录，可将同一主题消息归类到同一个目录。生产者向topic中写入数据时的详情如下：

  
  
  log_anatomy.png

• partition：一个topic可以有1到n个分区，可以在创建topic时，指定topic的分区数，单机版的kafka的topic默认只有一个分区。每个分区中的消息的顺序是有序、且不可改变的。

• offset：分区中的每条消息被标上一个有序的数字，这个数字就是offset。offset可以用来标识每条消息的位置。

  
  
  offset

• consumer group， 多个Consumer组成的一个分组，应用可以并发的消费一个topic下多个分区中的消息，并发的消费者数量由topic的分区数决定。如下图的topic中包含的4个分区，被Consumer Group A和Consumner GroupB同时消费。同一组中多个Consumer之间是竞争关系，共享一个offset，一条消息被一个Consumer消费之后就不能再被另一个Consumer消费。不同组的Consumer之间，各自有自己的offset，假设GroupA此时的offset为10，而GroupB的offset可能是5，二者之间不相互影响。

consumer-groups

对于同一个topic，同一组的Consumer同时消费，可以实现负载均衡，不同组的Consumer，可以实现对消息的重复消费。

• high-level API和low-level API。Kafka提供了一个high-level的Consumer API，它可以实现consumer group和自动容错，但是不能支持一些更复杂的使用场景，同也提供了一套simple的low-level API的Consumer API，提供更全面、更细粒度的控制，但是这种Consumer需要开发者自己设计容错机制。

Kafka Python实践

kafka的python包主要有kafka(pip install kafka)和kafka-python(pip install kafka-python)两种，建议使用kafka-python，kafka-python的API更丰富一些，下面的实践代码基于kafka-python。

1 简单的生产和消费

本文中servers的地址为：

servers = ['192.168.2.152:9092', '192.168.2.153:9092', '192.168.2.154:9092']

生产者

from kafka import KafkaProducer
def producer_message():
    producer = KafkaProducer(bootstrap_servers=servers)
    for i in range(100):
        msg = "some_message_bytes " + str(i)
        producer.send('kafka-topic', bytes(msg.encode("utf-8")))
        print(msg)

消费者

from kafka import KafkaConsumer
from kafka import TopicPartition

# 消费方式1，默认消费所有分区的消息
def consumer_message1():
    consumer = KafkaConsumer('kafka-topic', 
                             ootstrap_servers=servers, 
                             group_id="kafka-group-id")
    # consumer = KafkaConsumer('kafka-topic', bootstrap_servers=servers)
    for msg in consumer:
        print(msg)

# 消费方式2, 指定消费分区
def consumer_message2():
    consumer = KafkaConsumer(bootstrap_servers=servers, 
                             group_id="kafka-group-id")
    consumer.assign([TopicPartition('kafka-topic', 0)])
    for msg in consumer:
        print(msg)

# 消费方式3，手动commit，生产中建议使用这种方式
def consumer_message3():
    consumer = KafkaConsumer(bootstrap_servers=servers,
                             consumer_timeout_ms=1000,
                             group_id="kafka-group-id",
                             enable_auto_commit=False)
    consumer.assign([TopicPartition('kafka-topic', 0)])
    for msg in consumer:
        print(msg)
        consumer.commit()

2 简单的管理接口

获取所有的topic

kafka-python获取所有的topic接口是在KafkaConsumer类中实现的。笔者觉得这个方法放到KafkaAdminClient中可能更合适。

from kafka import KafkaConsumer

# 获取topic列表以及topic的分区列表
def retrieve_topics():
    consumer = KafkaConsumer(bootstrap_servers=servers)
    print(consumer.topics())

# 获取topic的分区列表
def retrieve_partitions(topic):
    consumer = KafkaConsumer(bootstrap_servers=servers)
    print(consumer.partitions_for_topic(topic))

# 获取Consumer Group对应的分区的当前偏移量
def retrieve_partition_offset():
    consumer = KafkaConsumer(bootstrap_servers=servers,
                             group_id='kafka-group-id')
    tp = TopicPartition('kafka-topic', 0)
    consumer.assign([tp])
    print("starting offset is ", consumer.position(tp))

创建、删除topic

from kafka import KafkaAdminClient
from kafka.admin import NewTopic
from kafka.errors import TopicAlreadyExistsError

admin = KafkaAdminClient(bootstrap_servers=servers)

# 创建topic
def create_topic():
    try:
        new_topic = NewTopic("create-topic", 8, 3)
        admin.create_topics([new_topic])
    except TopicAlreadyExistsError as e:
        print(e.message)

# 删除topic
def delete_topic():
    admin.delete_topics(["create-topic"])

获取消费组信息

from kafka import KafkaAdminClient

admin = KafkaAdminClient(bootstrap_servers=servers)
# 获取消费组信息    
def get_consumer_group():
    # 显示所有的消费组
    print(admin.list_consumer_groups())

    # 显示消费组的offsets
    print(admin.list_consumer_group_offsets("kafka-group-id"))

获取topic配置信息

from kafka import KafkaAdminClient
from kafka.admin import ConfigResource, ConfigResourceType

admin = KafkaAdminClient(bootstrap_servers=servers)
# 获取topic的配置信息
def get_topic_config():
    resource_config = ConfigResource(ConfigResourceType.TOPIC, "create-topic")
    config_entries = admin.describe_configs([resource_config])
    print(config_entries.resources)

kafka-python KafkaAdminClient中的bug

kafka-python包中的kafka\admin\client.py的第335行，

for topic, error_code in getattr(response, "topic_errors", response.topic_error_codes):

修改为

for topic, error_code, *a in getattr(response, "topic_errors", response.topic_error_codes):

image.png

总结

• 对于同一个topic，不同组之间的Consumer各自维持一个offset，同一组内的Consumer公用同一个offset。
• 当一个topic需要被不同的应用消费时，这些应用应设置不同的group_id，从而各自维持一个自己的offset。
• 若组内有多个Consumer并发消费，最好创建topic时指定topic的分区数量，topic的分区数量决定了同时能有多少个Consumer并发消费。

参考

• Apache Kafka
• Kafka streaming
• Introducing the Kafka Consumer
• kafka-python API