快捷導(dǎo)航

python操作kafka實(shí)踐的示例代碼

更新時(shí)間：2019年06月19日 09:17:42 作者：Small_office

這篇文章主要介紹了python操作kafka實(shí)踐的示例代碼，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

1、先看最簡(jiǎn)單的場(chǎng)景，生產(chǎn)者生產(chǎn)消息，消費(fèi)者接收消息，下面是生產(chǎn)者的簡(jiǎn)單代碼。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='xxxx:x')

msg_dict = {
  "sleep_time": 10,
  "db_config": {
    "database": "test_1",
    "host": "xxxx",
    "user": "root",
    "password": "root"
  },
  "table": "msg",
  "msg": "Hello World"
}
msg = json.dumps(msg_dict)
producer.send('test_rhj', msg, partition=0)
producer.close()

下面是消費(fèi)者的簡(jiǎn)單代碼：

from kafka import KafkaConsumer

consumer = 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

下面是結(jié)果：

2、如果想要完成負(fù)載均衡，就需要知道kafka的分區(qū)機(jī)制，同一個(gè)主題，可以為其分區(qū)，在生產(chǎn)者不指定分區(qū)的情況，kafka會(huì)將多個(gè)消息分發(fā)到不同的分區(qū)，消費(fèi)者訂閱時(shí)候如果不指定服務(wù)組，會(huì)收到所有分區(qū)的消息，如果指定了服務(wù)組，則同一服務(wù)組的消費(fèi)者會(huì)消費(fèi)不同的分區(qū)，如果2個(gè)分區(qū)兩個(gè)消費(fèi)者的消費(fèi)者組消費(fèi)，則，每個(gè)消費(fèi)者消費(fèi)一個(gè)分區(qū)，如果有三個(gè)消費(fèi)者的服務(wù)組，則會(huì)出現(xiàn)一個(gè)消費(fèi)者消費(fèi)不到數(shù)據(jù)；如果想要消費(fèi)同一分區(qū)，則需要用不同的服務(wù)組。以此為原理，我們對(duì)消費(fèi)者做如下修改：

from kafka import KafkaConsumer

consumer = 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

然后我們開(kāi)兩個(gè)消費(fèi)者進(jìn)行消費(fèi)，生產(chǎn)者分別往0分區(qū)和1分區(qū)發(fā)消息結(jié)果如下，可以看到，一個(gè)消費(fèi)者只能消費(fèi)0分區(qū)，另一個(gè)只能消費(fèi)1分區(qū)：

3、kafka提供了偏移量的概念，允許消費(fèi)者根據(jù)偏移量消費(fèi)之前遺漏的內(nèi)容，這基于kafka名義上的全量存儲(chǔ)，可以保留大量的歷史數(shù)據(jù)，歷史保存時(shí)間是可配置的，一般是7天，如果偏移量定位到了已刪除的位置那也會(huì)有問(wèn)題，但是這種情況可能很?。幻總€(gè)保存的數(shù)據(jù)文件都是以偏移量命名的，當(dāng)前要查的偏移量減去文件名就是數(shù)據(jù)在該文件的相對(duì)位置。要指定偏移量消費(fèi)數(shù)據(jù)，需要指定該消費(fèi)者要消費(fèi)的分區(qū)，否則代碼會(huì)找不到分區(qū)而無(wú)法消費(fèi)，代碼如下：

from kafka import KafkaConsumer
from kafka.structs import TopicPartition

consumer = 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主題的分區(qū)信息
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

因?yàn)橹付ǖ谋阋肆繛?，所以從一開(kāi)始插入的數(shù)據(jù)都可以查到，而且因?yàn)橹付朔謪^(qū)，指定的分區(qū)結(jié)果都可以消費(fèi)，結(jié)果如下：

4、有時(shí)候，我們并不需要實(shí)時(shí)獲取數(shù)據(jù)，因?yàn)檫@樣可能會(huì)造成性能瓶頸，我們只需要定時(shí)去獲取隊(duì)列里的數(shù)據(jù)然后批量處理就可以，這種情況，我們可以選擇主動(dòng)拉取數(shù)據(jù)

from kafka import KafkaConsumer
import time

consumer = 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 += 1
  print '--------poll index is %s----------' % index

結(jié)果如下，可以看到，每次拉取到的都是前面生產(chǎn)的數(shù)據(jù)，可能是多條的列表，也可能沒(méi)有數(shù)據(jù)，如果沒(méi)有數(shù)據(jù)，則拉取到的為空：