pip install confluent-kafka==1.9.2

kafka_setting = {
    'sasl_plain_username': 'XXX',   #如果是默認(rèn)接入點(diǎn)實(shí)例，請刪除該配置。
    'sasl_plain_password': 'XXX',   #如果是默認(rèn)接入點(diǎn)實(shí)例，請刪除該配置。
    'bootstrap_servers': '[xxx,xxx,xxx]',
    'topic_name': 'XXX',
    'group_name': 'XXX'
}

# -*- coding: utf-8 -*-
 
import json
import json
import msgpack
from loguru import logger
from kafka import KafkaProducer
from kafka.errors import KafkaError
 
def kfk_produce_1():
    """
        發(fā)送 json 格式數(shù)據(jù)
    :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():
    """
        發(fā)送 string 格式數(shù)據(jù)
    :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 ( 默認(rèn)是異步發(fā)送 )
    future = producer.send('my-topic', b'raw_bytes')
 
    # Block for 'synchronous' sends
    try:
        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 offset
    print(record_metadata.topic)
    print(record_metadata.partition)
    print(record_metadata.offset)
 
    # produce keyed messages to enable hashed partitioning
    producer.send('my-topic', key=b'foo', value=b'bar')
 
    # encode objects via msgpack
    producer = KafkaProducer(value_serializer=msgpack.dumps)
    producer.send('msgpack-topic', {'key': 'value'})
 
    # produce json messages
    producer = KafkaProducer(value_serializer=lambda m: json.dumps(m).encode('ascii'))
    producer.send('json-topic', {'key': 'value'})
 
    # produce asynchronously
    for _ 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 callbacks
    producer.send('my-topic', b'raw_bytes').add_callback(on_send_success).add_errback(on_send_error)
 
    # block until all async messages are sent
    producer.flush()
 
    # configure multiple retries
    producer = 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 ( 沒有消息時(shí)，1s后停止消費(fèi) )
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, KafkaConsumer
 
 
class KafkaOperate(object):
 
    def __init__(self, bootstrap_servers=None):
        if not bootstrap_servers:
            raise Exception('bootstrap_servers is None')
 
        self.__bootstrap_servers = None
        if 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 = None
        self.kafka_consumer = None
 
        pass
 
    def __del__(self):
        pass
 
    def 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 = 0
        for 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):
        """
            如果想要多線程進(jìn)行消費(fèi)，可以設(shè)置 發(fā)往不通的 partition
            有多少個(gè) partition 就可以啟多少個(gè)線程同時(shí)進(jìn)行消費(fèi)，
        :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',  # 同一個(gè)key值，會被送至同一個(gè)分區(qū)
                partition=partition
            )
        else:
            self.kafka_producer.send(topic_name, data_dict)
        pass
 
    def 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 = 0
        for data_dict in data_dict_list:
            partition = partition if partition else count % partition_count
            self.__kfk_produce(topic_name=topic_name, data_dict=data_dict, partition=partition)
            if 0 == count % per_count:
                self.kafka_producer.flush()
            count += 1
        self.kafka_producer.flush()
        pass
 
    @staticmethod
    def 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'))
            )
 
    @staticmethod
    def 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, KafkaProducer
 
 
class KProducer:
    def __init__(self, bootstrap_servers, topic):
        """
        kafka 生產(chǎn)者
        :param bootstrap_servers: 地址
        :param topic:  topic
        """
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda m: json.dumps(m).encode('ascii'), )  # json 格式化發(fā)送的內(nèi)容
        self.topic = topic
 
    def sync_producer(self, data_li: list):
        """
        同步發(fā)送 數(shù)據(jù)
        :param data_li:  發(fā)送數(shù)據(jù)
        :return:
        """
        for data in data_li:
            future = self.producer.send(self.topic, data)
            record_metadata = future.get(timeout=10)  # 同步確認(rèn)消費(fèi)
            partition = record_metadata.partition  # 數(shù)據(jù)所在的分區(qū)
            offset = record_metadata.offset  # 數(shù)據(jù)所在分區(qū)的位置
            print('save success, partition: {}, offset: {}'.format(partition, offset))
 
    def asyn_producer(self,  data_li: list):
        """
        異步發(fā)送數(shù)據(jù)
        :param data_li:發(fā)送數(shù)據(jù)
        :return:
        """
        for data in data_li:
            self.producer.send(self.topic, data)
        self.producer.flush()  # 批量提交
 
    def asyn_producer_callback(self,  data_li: list):
        """
        異步發(fā)送數(shù)據(jù) + 發(fā)送狀態(tài)處理
        :param data_li:發(fā)送數(shù)據(jù)
        :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):
        """異步發(fā)送成功回調(diào)函數(shù)"""
        print('save success')
        return
 
    def send_error(self, *args, **kwargs):
        """異步發(fā)送錯誤回調(diào)函數(shù)"""
        print('save error')
        return
 
    def close_producer(self):
        try:
            self.producer.close()
        except:
            pass
 
if __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')
 
    # 同步發(fā)送
    kp.sync_producer(send_data_li)
 
    # 異步發(fā)送
    # kp.asyn_producer(send_data_li)
 
    # 異步+回調(diào)
    # kp.asyn_producer_callback(send_data_li)
    
    kp.close_producer()

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)

from kafka import KafkaConsumer
 
consumer = 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)

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)

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)

# coding:utf8
from kafka import KafkaConsumer
 
# 創(chuàng)建一個(gè)消費(fèi)者，指定了topic,group_id,bootstrap_servers
# group_id: 多個(gè)擁有相同group_id的消費(fèi)者被判定為一組，
#            一條數(shù)據(jù)記錄只會被同一個(gè)組中的一個(gè)消費(fèi)者消費(fèi)
# bootstrap_servers：kafka的節(jié)點(diǎn)，多個(gè)節(jié)點(diǎn)使用逗號分隔
# 這種方式只會獲取新產(chǎn)生的數(shù)據(jù)
 
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",  # 消費(fèi)組id
    enable_auto_commit=True,  # 每過一段時(shí)間自動提交所有已消費(fèi)的消息（在迭代時(shí)提交）
    auto_commit_interval_ms=5000,  # 自動提交的周期（毫秒）
)
 
consumer.subscribe(["my.topic"])  # 消息的主題，可以指定多個(gè)
 
for msg in consumer:  # 迭代器，等待下一條消息
    print(msg)  # 打印消息

# coding:utf-8
 
import os
import sys
import threading
from kafka import KafkaConsumer, TopicPartition, OffsetAndMetadata
from collections import OrderedDict
 
threads = []
 
class MyThread(threading.Thread):
    def __init__(self, thread_name, topic, partition):
        threading.Thread.__init__(self)
        self.thread_name = thread_name
        self.partition = partition
        self.topic = topic
 
    def 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） - 服務(wù)器為獲取請求而返回的最小數(shù)據(jù)量，否則請等待
    fetch_max_wait_ms（int） - 如果沒有足夠的數(shù)據(jù)立即滿足fetch_min_bytes給出的要求，服務(wù)器在回應(yīng)提取請求之前將阻塞的最大時(shí)間量（以毫秒為單位）
    fetch_max_bytes（int） - 服務(wù)器應(yīng)為獲取請求返回的最大數(shù)據(jù)量。這不是絕對最大值，如果獲取的第一個(gè)非空分區(qū)中的第一條消息大于此值，
                則仍將返回消息以確保消費(fèi)者可以取得進(jìn)展。注意：使用者并行執(zhí)行對多個(gè)代理的提取，因此內(nèi)存使用將取決于包含該主題分區(qū)的代理的數(shù)量。
                支持的Kafka版本> = 0.10.1.0。默認(rèn)值：52428800（50 MB）。
    enable_auto_commit（bool） - 如果為True，則消費(fèi)者的偏移量將在后臺定期提交。默認(rèn)值：True。
    max_poll_records（int） - 單次調(diào)用中返回的最大記錄數(shù)poll()。默認(rèn)值：500
    max_poll_interval_ms（int） - poll()使用使用者組管理時(shí)的調(diào)用之間的最大延遲 。這為消費(fèi)者在獲取更多記錄之前可以閑置的時(shí)間量設(shè)置了上限。
                  如果 poll()在此超時(shí)到期之前未調(diào)用，則認(rèn)為使用者失敗，并且該組將重新平衡以便將分區(qū)重新分配給另一個(gè)成員。默認(rèn)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,  # 30s
        request_timeout_ms=305000,
        # consumer_timeout_ms=1,
        # max_poll_records=5000,
    )
    # 設(shè)置topic partition
    tp = TopicPartition(topic, partition)
    # 分配該消費(fèi)者的TopicPartition，也就是topic和partition，
    # 根據(jù)參數(shù)，每個(gè)線程消費(fèi)者消費(fèi)一個(gè)分區(qū)
    consumer_1.assign([tp])
    # 獲取上次消費(fèi)的最大偏移量
    offset = consumer_1.end_offsets([tp])[tp]
    print(thread_name, tp, offset)
 
    # 設(shè)置消費(fèi)的偏移量
    consumer_1.seek(tp, offset)
 
    print(u"程序首次運(yùn)行\(zhòng)t線程:", thread_name, u"分區(qū):", partition, u"偏移量:", offset, u"\t開始消費(fèi)...")
 
    num = 0  # 記錄該消費(fèi)者消費(fèi)次數(shù)
    while True:
        msg = consumer_1.poll(timeout_ms=60000)
        end_offset = consumer_1.end_offsets([tp])[tp]
        '''可以自己記錄控制消費(fèi)'''
        print(u'已保存的偏移量', consumer_1.committed(tp), u'最新偏移量，', end_offset)
        if len(msg) > 0:
            print(u"線程:", thread_name, u"分區(qū):", partition, u"最大偏移量:", end_offset, u"有無數(shù)據(jù),", len(msg))
 
            lines = 0
            for data in msg.values():
                for line in data:
                    print(line)
                    lines += 1
                '''
                do something
                '''
            # 線程此批次消息條數(shù)
 
            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:
                    # 系統(tǒng)退出？這個(gè)還沒試
                    os.exit()
                    '''
                    sys.exit()  只能退出該線程，也就是說其它兩個(gè)線程正常運(yùn)行，主程序不退出
                    '''
            else:
                os.exit()
        else:
            print(thread_name, '沒有數(shù)據(jù)')
        num += 1
        print(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")

# 立刻發(fā)送所有數(shù)據(jù)并等待發(fā)送完畢
producer.flush()
 
# 讀取下一條消息
next(consumer)
 
# 手動提交所有已消費(fèi)的消息
consumer.commit()
 
# 手動提交指定的消息
consumer.commit([TopicPartition(my_topic, msg.offset)])

import json
import traceback
from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import kafka_errors
 
def producer_demo():
    # 假設(shè)生產(chǎn)的消息為鍵值對（不是一定要鍵值對），且序列化方式為json
    producer = KafkaProducer(
        bootstrap_servers=['localhost:9092'],
        key_serializer=lambda k: json.dumps(k).encode(),
        value_serializer=lambda v: json.dumps(v).encode())
    # 發(fā)送三條消息
    for i in range(0, 3):
        future = producer.send(
            'kafka_demo',
            key='count_num',  # 同一個(gè)key值，會被送至同一個(gè)分區(qū)
            value=str(i),
            partition=1  # 向分區(qū)1發(fā)送消息
        )
        print("send {}".format(str(i)))
        try:
            future.get(timeout=10)  # 監(jiān)控是否發(fā)送成功           
        except kafka_errors:  # 發(fā)送失敗拋出kafka_errors
            traceback.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())}"
        )

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()

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