Python Kafka客户端confluent-kafka学习总结
实践环境Python 3.6.2
confluent-kafka 2.2.0
confluent-kafka简介
Confluent在GitHub上开发和维护的confluent-kafka-python，Apache Kafka®的一个python客户端，提供了一个与所有brokers>=v0.8的kafka 、Confluent Cloud和Confluent Platform兼容的高阶级生产者、消费者和AdminClient。
confluent-kafka安装
pip install confluent-kafka代码实践
Kafka生产者
from confluent_kafka import Producer
import socket
def acked(err, msg):
if err is not None:
print("Failed to deliver message: %s: %s" % msg.value(), str(err))
else:
print("Message produced: %s" % msg.value())
if __name__ == '__main__':
topic_name = 'FREE_TOPIC_FOR_TEST'
### 初始化Producer (针对本地运行的Kafka，即不在Confluent云平台上运行的Kafka）
conf = {'bootstrap.servers': '100.81.xxx.xx:9092,100.81.xxx.xx:9092',
'client.id': socket.gethostname()}
producer = Producer(conf)
### 异步写 kafka
# 给kafka发送消息--异步请求
producer.produce(topic_name, key="key", value="value")
# 添加回调函数
producer.produce(topic_name, value="test value", callback=acked)
# 最多等待事件1秒钟。等待期间，如果消息被确认，即成功写入kafka中，将调用回调 callback指定方法 acked
producer.poll(1)
### 同步写kafka
producer.produce(topic_name, key="key", value="new msg")
producer.flush()说明：
produce方法
producer.produce(topic, key="key", value="value", callback=None) # 给kafka发送消息
[*]topic kafka主题，如果主题不存在，则将自动创建
[*]key 可选
[*]value需要发送的消息，可以为None
[*]callback 回调函数。
product调用为异步请求，所以调用后立即完成，且不会返回值。如果由于librdkafka的本地生产队列已满而导致消息无法入队，则会引发KafkaException。
如果要接收发送是否成功或失败的通知，可以传递callback参数，该参数值可以是任何可调用的，例如lambda、函数、绑定方法或可调用对象。尽管produce()方法会立即将消息加入队列以进行批处理、压缩并传输到代理，但在调用poll()之前，不会传播任何传递通知事件。
flush方法
flush()方法用于同步写kafka。这通常是个坏主意，因为它有效地将吞吐量限制在broker往返时间内，但在某些情况下可能是合理的。
通常，应该在关闭生产者之前调用flush()，以确保所有未完成的/排队的/in-flight的消息都被传递。
Kafka消费者
import time
from confluent_kafka import Consumer
from confluent_kafka import KafkaException, KafkaError
running = True
def msg_process(msg):
value = msg.value()
if value:
value = value.decode('utf-8') # 假设消息可采用 utf-8解码
return {
'topic': msg.topic(),
'partition': msg.partition(),
'offset': msg.offset(),
'value': value
}
def consume_loop(consumer, topics):
global running
try:
consumer.subscribe(topics) # 订阅主题
while running:
msg = consumer.poll(timeout=10.0)
if msg is None:
time.sleep(0.1)
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
# End of partition event
print('%% %s [%d] reached end at offset %d\n' %
(msg.topic(), msg.partition(), msg.offset()))
elif msg.error():
raise KafkaException(msg.error())
else:
res = msg_process(msg)
try:
result = '{' + '"topic": "{topic}", "partition": {partition}, "offset": {offset}, "value": {value}'.format(**res) + '}\n'
except Exception:
result = '{' + '"topic": "{topic}", "partition": {partition}, "offset": {offset}, "value": "{value}"'.format(**res) + '}\n'
print(result)
finally:
# 关闭消费者以提交最后的偏移量
consumer.close()
if __name__ == '__main__':
topic_name = 'FREE_TOPIC_FOR_TEST_1'
# 初始化消费者
conf = {'bootstrap.servers': '100.81.xxx.xx:9092,100.81.xxx.xx:9092',
'group.id': 'custom_consumer',
'enable.auto.commit': 'true',
'auto.offset.reset': 'smallest',
}
consumer = Consumer(conf)
consume_loop(consumer, ) # 可以指定多个主题说明：
初始化消费者配置字典说明
conf = {'bootstrap.servers': 'host1:9092,host2:9092',
'group.id': 'foo',
'enable.auto.commit': 'false',
'auto.offset.reset': 'smallest'}说明：
[*]group.id属性是必需的，设置当前消费者归属的消费组，可以是事先不存在的消费组。
[*]auto.offset.reset属性指定针对当前消费组，在分区没有提交偏移量或提交偏移量无效（可能是由于日志截断）的情况下，消费者应该从哪个偏移量开始读取。可选值：
[*]'smallest'如果针对当前消费组，分区未提交offset，则从头开始消费，否则从已提交的offset 开始消费(即读取上次提交offset之后生产的数据)。
[*]'largest' 如果针对当前消费组，分区未提交offset，则读取新生产的数据(在启动该消费者之后才生产的数据)，不会读取之前的数据，否则从已提交的offset 开始消费，同smallest
[*]'earliest' 同 'smallest'
[*]'latest' 同 'largest'
kafka-0.10.1.X 版本之前：auto.offset.reset 的值为smallest和largest (offest保存在zk中)
kafka-0.10.1.X版本之后：auto.offset.reset 的值更改为 earliest, latest (offest保存在kafka的一个特殊的topic名为:__consumer_offsets里面)
[*]enable.auto.commit设置是否允许自动提交偏移量，默认为'true'，即允许。
一个典型的Kafka消费者应用程序以循环消费为中心，该循环重复调用poll方法来逐条检索消费者在后台高效预取的记录。例中poll超时被硬编码为1秒。如果在此期间没有收到任何记录，则Consumer.poll()将返回一个空记录集。
注意，在使用完Consumer之后，应该始终调用Consumer.close()，以确保活动套接字处于关闭状态，并清理内部状态。此外，还将立即触发组再均衡(group rebalance)，以确保消费者拥有的任何分区都被重新分配给组中的另一个成员。如果未正确关闭，broker将仅在会话超时到期后才触发再均衡。
同步提交
手动提交偏移量的最简单、最可靠的方法是为Consumer.commit()调用设置asynchronous参数，与此同时设置构建消费者对象参数配置'enable.auto.commit'为'false'。
MIN_COMMIT_COUNT = 10
def consume_loop(consumer, topics):
try:
consumer.subscribe(topics)
msg_count = 0
while running:
msg = consumer.poll(timeout=1.0)
if msg is None: continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
# End of partition event
print('%% %s [%d] reached end at offset %d\n' %
(msg.topic(), msg.partition(), msg.offset()))
elif msg.error():
raise KafkaException(msg.error())
else:
msg_process(msg)
msg_count += 1
if msg_count % MIN_COMMIT_COUNT == 0:
consumer.commit(asynchronous=False)
finally:
# 关闭消费者以提交最后的偏移量
consumer.close()在本例中，每消费MIN_COMMIT_COUNT 消息都会触发一次同步提交。asynchronous标志控制此调用是否为异步调用，默认为False，即同步 。您还可以在超时到期时触发提交，以确保定期更新提交的位置。
消息投递保证
在前面的示例中，由于提交在消息处理之后，所以获得了“至少一次(at least once)”投递。然而，通过更改提交偏移和处理消息的顺序，可获得“最多一次(at most once)”投递，但必须小心提交失败。
说明：
[*]最多一次(at most once)：消息可能丢失也可能被处理，但最多只会处理一次。因为当提交offset后，处理消息过程中出错导致消息处理失败，或者消费者down掉，导致消息不被处理。
[*]至少一次(at least once)：消息不会丢失，但可能被处理多次。先获取消息，然后处理消息，最后提交offset，提交offset时，可能会因为网络超时，消费者down掉等，导致提交偏移量失败的情况，所以，会导致重复消费消息的情况，进而导致多次处理消息。
def consume_loop(consumer, topics):
try:
consumer.subscribe(topics)
while running:
msg = consumer.poll(timeout=1.0)
if msg is None: continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
# End of partition event
print('%% %s [%d] reached end at offset %d\n' %
(msg.topic(), msg.partition(), msg.offset()))
elif msg.error():
raise KafkaException(msg.error())
else:
consumer.commit(asynchronous=False)
msg_process(msg)
finally:
# 关闭消费者以提交最后的偏移量
consumer.close()简单起见，在本例中，在处理消息之前使用Consumer.commit()。在实践中，对每条消息都进行提交会产生大量开销。更好的方法是收集一批消息，执行同步提交，然后只有在提交成功的情况下才处理消息。
异步提交
def consume_loop(consumer, topics):
try:
consumer.subscribe(topics)
msg_count = 0
while running:
msg = consumer.poll(timeout=1.0)
if msg is None: continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
# End of partition event
print('%% %s [%d] reached end at offset %d\n' %
(msg.topic(), msg.partition(), msg.offset()))
elif msg.error():
raise KafkaException(msg.error())
else:
msg_process(msg)
msg_count += 1
if msg_count % MIN_COMMIT_COUNT == 0:
consumer.commit(asynchronous=True)
finally:
consumer.close()本例在前面示例的基础上，将commit() 的asynchronous 参数改成True，消费者将使用异步提交发送请求并立即返回
API提供了一个callback，当提交成功或失败时会调用该callback。 commit callback回调可以是任何可调用的，并且可以作为配置参数传递给消费者构造函数。
from confluent_kafka import Consumer
def commit_completed(err, partitions):
if err:
print(str(err))
else:
print("Committed partition offsets: " + str(partitions))
conf = {'bootstrap.servers': "host1:9092,host2:9092",
'group.id': "foo",
'auto.offset.reset': 'smallest',
'on_commit': commit_completed}
consumer = Consumer(conf)参考连接
https://docs.confluent.io/kafka-clients/python/current/overview.html#initialization
https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html
来源:https://www.cnblogs.com/shouke/p/17814004.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！
页:
[1]