kafka-python：https://github.com/dpkp/kafka-python

kafka-python 文檔：https://kafka-python.readthedocs.io/en/master/apidoc/modules.html

kafka 官方文檔：http://kafka.apache.org/documentation.html

Python 操作 Kafka 的通俗總結（kafka-python）：https://zhuanlan.zhihu.com/p/279784873

譯：Kafka 和 Unix 管道的示例：http://zqhxuyuan.github.io/2016/01/05/2016-01-05-Kafka-Unix/

一、基本概念

• Topic：一組消息數據的標記符；
• Producer：生產者，用于生產數據，可將生產后的消息送入指定的 Topic；
• Consumer：消費者，獲取數據，可消費指定的 Topic 里面的數據
• Group：消費者組，同一個 group 可以有多個消費者，一條消息在一個 group 中，只會被一個消費者 獲取；
• Partition：分區，為了保證 kafka 的吞吐量，一個 Topic 可以設置多個分區。同一分區只能被一個消費者訂閱。

二、安裝 kafka-python

pip 命令：pip install kafka-python

三、生產者（Producer）與 消費者（Consumer）

生產者 示例：

# -*- 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

?消費者 示例：

# -*- 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' )

簡單封裝：

# -*- 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

示例：

# -*- 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))

總結

以上是生活随笔為你收集整理的Python 操作 Kafka --- kafka-python的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。