当前位置：首页 > news >正文

python操作kafka

news 来源：原创 2024/9/20 17:34:35

一、参考阿里云的官方链接：

使用Python SDK接入Kafka收发消息_云消息队列 Kafka 版(Kafka)-阿里云帮助中心

二、安装python环境

三、添加python依赖库

pip install confluent-kafka==1.9.2

四、新建一个setting.py文件配置信息

kafka_setting = {'sasl_plain_username': 'XXX',   #如果是默认接入点实例，请删除该配置。'sasl_plain_password': 'XXX',   #如果是默认接入点实例，请删除该配置。'bootstrap_servers': '[xxx,xxx,xxx]','topic_name': 'XXX','group_name': 'XXX'
}

五、生产者和消费者

5.1 生产者示例：

# -*- coding: utf-8 -*-import json
import json
import msgpack
from loguru import logger
from kafka import KafkaProducer
from kafka.errors import KafkaErrordef kfk_produce_1():"""发送 json 格式数据:return:"""producer = KafkaProducer(bootstrap_servers='ip:9092',value_serializer=lambda v: json.dumps(v).encode('utf-8'))producer.send('test_topic', {'key1': 'value1'})def kfk_produce_2():"""发送 string 格式数据:return:"""producer = KafkaProducer(bootstrap_servers='xxxx:x')data_dict = {"name": 'king','age': 100,"msg": "Hello World"}msg = json.dumps(data_dict)producer.send('test_topic', msg, partition=0)producer.close()def kfk_produce_3():producer = KafkaProducer(bootstrap_servers=['broker1:1234'])# Asynchronous by default ( 默认是异步发送 )future = producer.send('my-topic', b'raw_bytes')# Block for 'synchronous' sendstry:record_metadata = future.get(timeout=10)except KafkaError:# Decide what to do if produce request failed...logger.error(KafkaError)pass# Successful result returns assigned partition and offsetprint(record_metadata.topic)print(record_metadata.partition)print(record_metadata.offset)# produce keyed messages to enable hashed partitioningproducer.send('my-topic', key=b'foo', value=b'bar')# encode objects via msgpackproducer = KafkaProducer(value_serializer=msgpack.dumps)producer.send('msgpack-topic', {'key': 'value'})# produce json messagesproducer = KafkaProducer(value_serializer=lambda m: json.dumps(m).encode('ascii'))producer.send('json-topic', {'key': 'value'})# produce asynchronouslyfor _ in range(100):producer.send('my-topic', b'msg')def on_send_success(record_metadata=None):print(record_metadata.topic)print(record_metadata.partition)print(record_metadata.offset)def on_send_error(excp=None):logger.error('I am an errback', exc_info=excp)# handle exception# produce asynchronously with callbacksproducer.send('my-topic', b'raw_bytes').add_callback(on_send_success).add_errback(on_send_error)# block until all async messages are sentproducer.flush()# configure multiple retriesproducer = KafkaProducer(retries=5)if __name__ == '__main__':kfk_produce_1()kfk_produce_2()pass

5.2 消费者示例：

# -*- coding: utf-8 -*-import json
import msgpack
from kafka import KafkaConsumer# 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')`info = f'{message.topic}:{message.partition}:{message.offset}: key={message.key}, value={message.value}'print(info)# 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 ( 没有消息时，1s后停止消费 )
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'
)

5.3 简单封装：

# -*- coding: utf-8 -*-import time
import json
import ujson
import random
from loguru import logger
from kafka import KafkaProducer, KafkaConsumerclass KafkaOperate(object):def __init__(self, bootstrap_servers=None):if not bootstrap_servers:raise Exception('bootstrap_servers is None')self.__bootstrap_servers = Noneif isinstance(bootstrap_servers, str):ip_port_string = bootstrap_servers.strip()if ',' in ip_port_string:self.__bootstrap_servers = ip_port_string.replace(' ', '').split(',')else:self.__bootstrap_servers = [ip_port_string]self.kafka_producer = Noneself.kafka_consumer = Nonepassdef __del__(self):passdef kfk_consume(self, topic_name=None, group_id='my_group'):if not self.kafka_consumer:self.kafka_consumer = KafkaConsumer(topic_name, group_id=group_id,bootstrap_servers=self.__bootstrap_servers,auto_offset_reset='earliest',)count = 0for msg in self.kafka_consumer:count += 1# message value and key are raw bytes -- decode if necessary!# e.g., for unicode: `message.value.decode('utf-8')`info = f'[{count}] {msg.topic}:{msg.partition}:{msg.offset}: key={msg.key}, value={msg.value.decode("utf-8")}'logger.info(info)time.sleep(1)def __kfk_produce(self, topic_name=None, data_dict=None, partition=None):"""如果想要多线程进行消费，可以设置 发往不通的 partition有多少个 partition 就可以启多少个线程同时进行消费，:param topic_name::param data_dict::param partition::return:"""if not self.kafka_producer:self.kafka_producer = KafkaProducer(bootstrap_servers=self.__bootstrap_servers,client_id='my_group',value_serializer=lambda v: json.dumps(v).encode('utf-8'))# data_dict = {#     "name": 'king',#     'age': 100,#     "msg": "Hello World"# }if partition:self.kafka_producer.send(topic=topic_name, value=data_dict,# key='count_num',  # 同一个key值，会被送至同一个分区partition=partition)else:self.kafka_producer.send(topic_name, data_dict)passdef kfk_produce_one(self, topic_name=None, data_dict=None, partition=None, partition_count=1):partition = partition if partition else random.randint(0, partition_count-1)self.__kfk_produce(topic_name=topic_name, data_dict=data_dict, partition=partition)self.kafka_producer.flush()def kfk_produce_many(self, topic_name=None, data_dict_list=None, partition=None, partition_count=1, per_count=100):count = 0for data_dict in data_dict_list:partition = partition if partition else count % partition_countself.__kfk_produce(topic_name=topic_name, data_dict=data_dict, partition=partition)if 0 == count % per_count:self.kafka_producer.flush()count += 1self.kafka_producer.flush()pass@staticmethoddef get_consumer(group_id: str, bootstrap_servers: list, topic: str, enable_auto_commit=True) -> KafkaConsumer:topics = tuple([x.strip() for x in topic.split(',') if x.strip()])if enable_auto_commit:return KafkaConsumer(*topics,group_id=group_id,bootstrap_servers=bootstrap_servers,auto_offset_reset='earliest',# fetch_max_bytes=FETCH_MAX_BYTES,# connections_max_idle_ms=CONNECTIONS_MAX_IDLE_MS,# max_poll_interval_ms=KAFKA_MAX_POLL_INTERVAL_MS,# session_timeout_ms=SESSION_TIMEOUT_MS,# max_poll_records=KAFKA_MAX_POLL_RECORDS,# request_timeout_ms=REQUEST_TIMEOUT_MS,# auto_commit_interval_ms=AUTO_COMMIT_INTERVAL_MS,value_deserializer=lambda m: ujson.loads(m.decode('utf-8')))else:return KafkaConsumer(*topics,group_id=group_id,bootstrap_servers=bootstrap_servers,auto_offset_reset='earliest',# fetch_max_bytes=FETCH_MAX_BYTES,# connections_max_idle_ms=CONNECTIONS_MAX_IDLE_MS,# max_poll_interval_ms=KAFKA_MAX_POLL_INTERVAL_MS,# session_timeout_ms=SESSION_TIMEOUT_MS,# max_poll_records=KAFKA_MAX_POLL_RECORDS,# request_timeout_ms=REQUEST_TIMEOUT_MS,enable_auto_commit=enable_auto_commit,value_deserializer=lambda m: ujson.loads(m.decode('utf-8')))@staticmethoddef get_producer(bootstrap_servers: list):return KafkaProducer(bootstrap_servers=bootstrap_servers, retries=5)if __name__ == '__main__':bs = '10.10.10.10:9092'kafka_op = KafkaOperate(bootstrap_servers=bs)kafka_op.kfk_consume(topic_name='001_test')pass

5.4 示例：

# -*- coding:utf-8 -*-import json
from kafka import KafkaConsumer, KafkaProducerclass KProducer:def __init__(self, bootstrap_servers, topic):"""kafka 生产者:param bootstrap_servers: 地址:param topic:  topic"""self.producer = KafkaProducer(bootstrap_servers=bootstrap_servers,value_serializer=lambda m: json.dumps(m).encode('ascii'), )  # json 格式化发送的内容self.topic = topicdef sync_producer(self, data_li: list):"""同步发送 数据:param data_li:  发送数据:return:"""for data in data_li:future = self.producer.send(self.topic, data)record_metadata = future.get(timeout=10)  # 同步确认消费partition = record_metadata.partition  # 数据所在的分区offset = record_metadata.offset  # 数据所在分区的位置print('save success, partition: {}, offset: {}'.format(partition, offset))def asyn_producer(self,  data_li: list):"""异步发送数据:param data_li:发送数据:return:"""for data in data_li:self.producer.send(self.topic, data)self.producer.flush()  # 批量提交def asyn_producer_callback(self,  data_li: list):"""异步发送数据 + 发送状态处理:param data_li:发送数据:return:"""for data in data_li:self.producer.send(self.topic, data).add_callback(self.send_success).add_errback(self.send_error)self.producer.flush()  # 批量提交def send_success(self, *args, **kwargs):"""异步发送成功回调函数"""print('save success')returndef send_error(self, *args, **kwargs):"""异步发送错误回调函数"""print('save error')returndef close_producer(self):try:self.producer.close()except:passif __name__ == '__main__':send_data_li = [{"test": 1}, {"test": 2}]kp = KProducer(topic='topic', bootstrap_servers='127.0.0.1:9001,127.0.0.1:9002')# 同步发送kp.sync_producer(send_data_li)# 异步发送# kp.asyn_producer(send_data_li)# 异步+回调# kp.asyn_producer_callback(send_data_li)kp.close_producer()

KafkaConsumer 的构造参数：

*topics ，要订阅的主题
auto_offset_reset：总共3种值：earliest 、latest、none
  earliest ：
  当各分区下有已提交的 offset 时，从提交的 offset 开始消费；
无提交的 offset时，从头开始消费
  latest ：
  当各分区下有已提交的offset时，从提交的offset开始消费；
无提交的offset时，消费新产生的该分区下的数据
  none ：
  topic 各分区都存在已提交的 offset 时，从 offset 后开始消费；
只要有一个分区不存在已提交的 offset，则抛出异常
bootstrap_servers ：kafka节点或节点的列表，不一定需要罗列所有的kafka节点。格式为： ‘host[:port]’ 。默认值是：localhost:9092
client_id (str) : 客户端id，默认值: ‘kafka-python-{version}’
group_id (str or None)：分组id
key_deserializer (callable) ：key反序列化函数
value_deserializer (callable)：value反序列化函数
fetch_min_bytes：服务器应每次返回的最小数据量
fetch_max_wait_ms (int)：服务器应每次返回的最大等待时间
fetch_max_bytes (int) ：服务器应每次返回的最大数据量
max_partition_fetch_bytes (int) ：
request_timeout_ms (int) retry_backoff_ms (int)
reconnect_backoff_ms (int)
reconnect_backoff_max_ms (int)
max_in_flight_requests_per_connection (int)
auto_offset_reset (str) enable_auto_commit (bool)
auto_commit_interval_ms (int)
default_offset_commit_callback (callable)
check_crcs (bool)
metadata_max_age_ms (int)
partition_assignment_strategy (list)
max_poll_records (int)
max_poll_interval_ms (int)
session_timeout_ms (int)
heartbeat_interval_ms (int)
receive_buffer_bytes (int)
send_buffer_bytes (int)
socket_options (list)
consumer_timeout_ms (int)
skip_double_compressed_messages (bool)
security_protocol (str)
ssl_context (ssl.SSLContext)
ssl_check_hostname (bool)
ssl_cafile (str) –
ssl_certfile (str)
ssl_keyfile (str)
ssl_password (str)
ssl_crlfile (str)
api_version (tuple)

KafkaConsumer 的函数

assign(partitions)：手动为该消费者分配一个topic分区列表。
assignment()：获取当前分配给该消费者的topic分区。
beginning_offsets(partitions)：获取给定分区的第一个偏移量。
close(autocommit=True)：关闭消费者
commit(offsets=None)：提交偏移量，直到成功或错误为止。
commit_async(offsets=None, callback=None)：异步提交偏移量。
committed(partition)：获取给定分区的最后一个提交的偏移量。
end_offsets(partitions)：获取分区的最大偏移量
highwater(partition)：分区最大的偏移量
metrics(raw=False)：返回消费者性能指标
next（）：返回下一条数据
offsets_for_times(timestamps)：根据时间戳获取分区偏移量
partitions_for_topic(topic)：返回topic的partition列表，返回一个set集合
pause(*partitions)：停止获取数据paused()：返回停止获取的分区poll(timeout_ms=0, max_records=None)：获取数据
position(partition)：获取分区的偏移量
resume(*partitions)：恢复抓取指定的分区
seek(partition, offset)：seek偏移量
seek_to_beginning(*partitions)：搜索最旧的偏移量
seek_to_end(*partitions)：搜索最近可用的偏移量
subscribe(topics=(), pattern=None, listener=None)：订阅topics
subscription()：返回当前消费者消费的所有topic
topics()：返回当前消费者消费的所有topic，返回的是unicode
unsubscribe()：取消订阅所有的topic

六、简单的消费者代码：

from kafka import KafkaConsumerconsumer = KafkaConsumer('test_rhj', bootstrap_servers=['xxxx:x'])
for msg in consumer:recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)print(recv)

七、kafka 的分区机制

如果想要完成负载均衡，就需要知道 kafka 的分区机制，

同一个主题 ( topic ) ，可以为其分区，
生产者在不指定分区的情况，kafka 会将多个消息分发到不同的分区，

消费者订阅时候

如果不指定服务组，会收到所有分区的消息，
如果指定了服务组，则同一服务组的消费者会消费不同的分区，
如果2个分区两个消费者的消费者组消费，则每个消费者消费一个分区，
如果有三个消费者的服务组，则会出现一个消费者消费不到数据；如果想要消费同一分区，则需要用不同的服务组。

以此为原理，我们对消费者做如下修改：

from kafka import KafkaConsumerconsumer = KafkaConsumer('test_rhj', group_id='123456', bootstrap_servers=['10.43.35.25:4531']
)
for msg in consumer:recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)print(recv)

开两个消费者进行消费，生产者分别往 0分区和 1分区发消息结果如下，可以看到，一个消费者只能消费0分区，另一个只能消费1分区：

八、偏移量

kafka 提供了 "偏移量" 的概念，允许消费者根据偏移量消费之前遗漏的内容，这基于 kafka 名义上的全量存储，可以保留大量的历史数据，历史保存时间是可配置的，一般是7天，如果偏移量定位到了已删除的位置那也会有问题，但是这种情况可能很小；每个保存的数据文件都是以偏移量命名的，当前要查的偏移量减去文件名就是数据在该文件的相对位置。要指定偏移量消费数据，需要指定该消费者要消费的分区，否则代码会找不到分区而无法消费，代码如下：

from kafka import KafkaConsumer
from kafka.structs import TopicPartitionconsumer = KafkaConsumer(group_id='123456', bootstrap_servers=['10.43.35.25:4531']
)
consumer.assign([TopicPartition(topic='test_rhj', partition=0),TopicPartition(topic='test_rhj', partition=1)]
)print(consumer.partitions_for_topic("test_rhj"))  # 获取test主题的分区信息
print(consumer.assignment())
print(consumer.beginning_offsets(consumer.assignment()))consumer.seek(TopicPartition(topic='test_rhj', partition=0), 0)
for msg in consumer:recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)print(recv)

因为指定的偏移量为 0，所以从一开始插入的数据都可以查到，而且因为指定了分区，指定的分区结果都可以消费，结果如下：

有时候，我们并不需要实时获取数据，因为这样可能会造成性能瓶颈，我们只需要定时去获取队列里的数据然后批量处理就可以，这种情况，我们可以选择主动拉取数据

from kafka import KafkaConsumer
import timeconsumer = KafkaConsumer(group_id='123456', bootstrap_servers=['10.43.35.25:4531'])
consumer.subscribe(topics=('test_rhj',))
index = 0
while True:msg = consumer.poll(timeout_ms=5)  # 从kafka获取消息print(msg)time.sleep(2)index += 1print('--------poll index is %s----------' % index)

结果如下，可以看到，每次拉取到的都是前面生产的数据，可能是多条的列表，也可能没有数据，如果没有数据，则拉取到的为空：

九、消费者示例

# coding:utf8
from kafka import KafkaConsumer# 创建一个消费者，指定了topic,group_id,bootstrap_servers
# group_id: 多个拥有相同group_id的消费者被判定为一组，
#            一条数据记录只会被同一个组中的一个消费者消费
# bootstrap_servers：kafka的节点，多个节点使用逗号分隔
# 这种方式只会获取新产生的数据bootstrap_server_list = ['192.168.70.221:19092','192.168.70.222:19092','192.168.70.223:19092'
]consumer = KafkaConsumer(# kafka 集群地址bootstrap_servers=','.join(bootstrap_server_list),group_id="my.group",  # 消费组idenable_auto_commit=True,  # 每过一段时间自动提交所有已消费的消息（在迭代时提交）auto_commit_interval_ms=5000,  # 自动提交的周期（毫秒）
)consumer.subscribe(["my.topic"])  # 消息的主题，可以指定多个for msg in consumer:  # 迭代器，等待下一条消息print(msg)  # 打印消息

十、多线程消费

# coding:utf-8import os
import sys
import threading
from kafka import KafkaConsumer, TopicPartition, OffsetAndMetadata
from collections import OrderedDictthreads = []class MyThread(threading.Thread):def __init__(self, thread_name, topic, partition):threading.Thread.__init__(self)self.thread_name = thread_nameself.partition = partitionself.topic = topicdef run(self):print("Starting " + self.name)consumer(self.thread_name, self.topic, self.partition)def stop(self):sys.exit()def consumer(thread_name, topic, partition):broker_list = 'ip1:9092,ip2:9092''''fetch_min_bytes（int） - 服务器为获取请求而返回的最小数据量，否则请等待fetch_max_wait_ms（int） - 如果没有足够的数据立即满足fetch_min_bytes给出的要求，服务器在回应提取请求之前将阻塞的最大时间量（以毫秒为单位）fetch_max_bytes（int） - 服务器应为获取请求返回的最大数据量。这不是绝对最大值，如果获取的第一个非空分区中的第一条消息大于此值，则仍将返回消息以确保消费者可以取得进展。注意：使用者并行执行对多个代理的提取，因此内存使用将取决于包含该主题分区的代理的数量。支持的Kafka版本> = 0.10.1.0。默认值：52428800（50 MB）。enable_auto_commit（bool） - 如果为True，则消费者的偏移量将在后台定期提交。默认值：True。max_poll_records（int） - 单次调用中返回的最大记录数poll()。默认值：500max_poll_interval_ms（int） - poll()使用使用者组管理时的调用之间的最大延迟 。这为消费者在获取更多记录之前可以闲置的时间量设置了上限。如果 poll()在此超时到期之前未调用，则认为使用者失败，并且该组将重新平衡以便将分区重新分配给另一个成员。默认300000'''consumer_1 = KafkaConsumer(bootstrap_servers=broker_list,group_id="test000001",client_id=thread_name,enable_auto_commit=False,fetch_min_bytes=1024 * 1024,  # 1M# fetch_max_bytes=1024 * 1024 * 1024 * 10,fetch_max_wait_ms=60000,  # 30srequest_timeout_ms=305000,# consumer_timeout_ms=1,# max_poll_records=5000,)# 设置topic partitiontp = TopicPartition(topic, partition)# 分配该消费者的TopicPartition，也就是topic和partition，# 根据参数，每个线程消费者消费一个分区consumer_1.assign([tp])# 获取上次消费的最大偏移量offset = consumer_1.end_offsets([tp])[tp]print(thread_name, tp, offset)# 设置消费的偏移量consumer_1.seek(tp, offset)print(u"程序首次运行\t线程:", thread_name, u"分区:", partition, u"偏移量:", offset, u"\t开始消费...")num = 0  # 记录该消费者消费次数while True:msg = consumer_1.poll(timeout_ms=60000)end_offset = consumer_1.end_offsets([tp])[tp]'''可以自己记录控制消费'''print(u'已保存的偏移量', consumer_1.committed(tp), u'最新偏移量，', end_offset)if len(msg) > 0:print(u"线程:", thread_name, u"分区:", partition, u"最大偏移量:", end_offset, u"有无数据,", len(msg))lines = 0for data in msg.values():for line in data:print(line)lines += 1'''do something'''# 线程此批次消息条数print(thread_name, "lines", lines)if True:# 可以自己保存在各topic, partition的偏移量# 手动提交偏移量 offsets格式：{TopicPartition:OffsetAndMetadata(offset_num,None)}consumer_1.commit(offsets={tp: (OffsetAndMetadata(end_offset, None))})if not 0:# 系统退出？这个还没试os.exit()'''sys.exit()  只能退出该线程，也就是说其它两个线程正常运行，主程序不退出'''else:os.exit()else:print(thread_name, '没有数据')num += 1print(thread_name, "第", num, "次")if __name__ == '__main__':try:t1 = MyThread("Thread-0", "test", 0)threads.append(t1)t2 = MyThread("Thread-1", "test", 1)threads.append(t2)t3 = MyThread("Thread-2", "test", 2)threads.append(t3)for t in threads:t.start()for t in threads:t.join()print("exit program with 0")except:print("Error: failed to run consumer program")

十一、高级用法（消费者）

从指定 offset 开始读取消息，被消费过的消息也可以被此方法读取

创建消费者

使用 assign 方法重置指定分区(partition)的读取偏移(fetch offset)的值

使用 seek 方法从指定的partition和offset开始读取数据

#encoding:utf8
from kafka import KafkaConsumer, TopicPartitionmy_topic = "my.topic" # 指定需要消费的主题consumer = KafkaConsumer(# kafka集群地址bootstrap_servers = "192.168.70.221:19092,192.168.70.222:19092", group_id = "my.group", # 消费组idenable_auto_commit = True, # 每过一段时间自动提交所有已消费的消息（在迭代时提交）auto_commit_interval_ms = 5000, # 自动提交的周期（毫秒）
)consumer.assign([TopicPartition(topic=my_topic, partition=0),TopicPartition(topic=my_topic, partition=1),TopicPartition(topic=my_topic, partition=2)
])# 指定起始 offset 为 12
consumer.seek(TopicPartition(topic=my_topic, partition=0), 12)# 可以注册多个分区，此分区从第一条消息开始接收
consumer.seek(TopicPartition(topic=my_topic, partition=1), 0) # 没有注册的分区上的消息不会被消费
# consumer.seek(TopicPartition(topic=my_topic, partition=2), 32)for msg in consumer: # 迭代器，等待下一条消息print msg # 打印消息

其他用法

# 立刻发送所有数据并等待发送完毕
producer.flush()# 读取下一条消息
next(consumer)# 手动提交所有已消费的消息
consumer.commit()# 手动提交指定的消息
consumer.commit([TopicPartition(my_topic, msg.offset)])

十二、生产者和消费者的 Demo

import json
import traceback
from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import kafka_errorsdef producer_demo():# 假设生产的消息为键值对（不是一定要键值对），且序列化方式为jsonproducer = KafkaProducer(bootstrap_servers=['localhost:9092'],key_serializer=lambda k: json.dumps(k).encode(),value_serializer=lambda v: json.dumps(v).encode())# 发送三条消息for i in range(0, 3):future = producer.send('kafka_demo',key='count_num',  # 同一个key值，会被送至同一个分区value=str(i),partition=1  # 向分区1发送消息)print("send {}".format(str(i)))try:future.get(timeout=10)  # 监控是否发送成功           except kafka_errors:  # 发送失败抛出kafka_errorstraceback.format_exc()def consumer_demo():consumer = KafkaConsumer('kafka_demo',bootstrap_servers=':9092',group_id='test')for message in consumer:print(f"receive, key: {json.loads(message.key.decode())}, "f"value: {json.loads(message.value.decode())}")

十三、消费者进阶操作

（1）初始化参数：

列举一些 KafkaConsumer 初始化时的重要参数：

group_id ：高并发量，则需要有多个消费者协作，消费进度，则由group_id统一。例如消费者A与消费者B，在初始化时使用同一个group_id。在进行消费时，一条消息被消费者A消费后，在kafka中会被标记，这条消息不会再被B消费（前提是A消费后正确commit）。
key_deserializer， value_deserializer ：与生产者中的参数一致，自动解析。
auto_offset_reset ：消费者启动的时刻，消息队列中或许已经有堆积的未消费消息，有时候需求是从上一次未消费的位置开始读（则该参数设置为 earliest ），有时候的需求为从当前时刻开始读之后产生的，之前产生的数据不再消费（则该参数设置为 latest ）。
enable_auto_commit， auto_commit_interval_ms ：是否自动commit，当前消费者消费完该数据后，需要commit，才可以将消费完的信息传回消息队列的控制中心。enable_auto_commit 设置为 True 后，消费者将自动 commit，并且两次 commit 的时间间隔为 auto_commit_interval_ms 。

（2）手动 commit

def consumer_demo():consumer = KafkaConsumer('kafka_demo', bootstrap_servers=':9092',group_id='test',enable_auto_commit=False)for message in consumer:print(f"receive, key: {json.loads(message.key.decode())}, "f"value: {json.loads(message.value.decode())}")consumer.commit()

（3）查看 kafka 堆积剩余量

在线环境中，需要保证消费者的消费速度大于生产者的生产速度，所以需要检测 kafka 中的剩余堆积量是在增加还是减小。可以用如下代码，观测队列消息剩余量：

consumer = KafkaConsumer(topic, **kwargs)
partitions = [TopicPartition(topic, p) for p in consumer.partitions_for_topic(topic)]print("start to cal offset:")# total
toff = consumer.end_offsets(partitions)
toff = [(key.partition, toff[key]) for key in toff.keys()]
toff.sort()
print("total offset: {}".format(str(toff)))# current
coff = [(x.partition, consumer.committed(x)) for x in partitions]
coff.sort()
print("current offset: {}".format(str(coff)))# 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
print("kafka left: {}".format(left_sum))