pip install --user kafka-python==1.4.3 

yum install snappy-devel
yum install lz4-devel
pip install python-snappy
pip install lz4

#!/usr/bin/env python
# coding : utf-8

from kafka import KafkaProducer
import json

def kafkaProducer():
  producer = KafkaProducer(bootstrap_servers='ip:9092',value_serializer=lambda v: json.dumps(v).encode('utf-8'))
  producer.send('world', {'key1': 'value1'})

if __name__ == '__main__':
  kafkaProducer()

from kafka import KafkaConsumer
from kafka.structs import TopicPartition
import time
import click
import ConfigParser
import json
import threading
import datetime
import sched


config = ConfigParser.ConfigParser()
config.read("amon.ini")

@click.group()
def cli():
  pass

@cli.command()
@click.option('--topic',type=str)
@click.option('--offset', type=click.Choice(['smallest', 'earliest', 'largest']))
@click.option("--group",type=str)
def client(topic,offset,group):
  click.echo(topic)
  consumer = KafkaConsumer(topic,
               bootstrap_servers=config.get("KAFKA", "Broker_Servers").split(","),
               group_id=group,
               auto_offset_reset=offset)
  for message in consumer:
    click.echo(message.value)
    # click.echo("%d:%d: key=%s value=%s" % (message.partition,
    #                      message.offset, message.key,
    #                      message.value))

if __name__ == '__main__':
  cli()

#coding:utf-8
import threading

import os
import sys
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ù)器為獲取請(qǐng)求而返回的最小數(shù)據(jù)量，否則請(qǐng)等待
  fetch_max_wait_ms（int） - 如果沒(méi)有足夠的數(shù)據(jù)立即滿足fetch_min_bytes給出的要求，服務(wù)器在回應(yīng)提取請(qǐng)求之前將阻塞的最大時(shí)間量（以毫秒為單位）
  fetch_max_bytes（int） - 服務(wù)器應(yīng)為獲取請(qǐng)求返回的最大數(shù)據(jù)量。這不是絕對(duì)最大值，如果獲取的第一個(gè)非空分區(qū)中的第一條消息大于此值，
              則仍將返回消息以確保消費(fèi)者可以取得進(jìn)展。注意：使用者并行執(zhí)行對(duì)多個(gè)代理的提取，因此內(nèi)存使用將取決于包含該主題分區(qū)的代理的數(shù)量。
              支持的Kafka版本> = 0.10.1.0。默認(rèn)值：52428800（50 MB）。
  enable_auto_commit（bool） - 如果為True，則消費(fèi)者的偏移量將在后臺(tá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 = 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.assign([tp])
  #獲取上次消費(fèi)的最大偏移量
  offset = consumer.end_offsets([tp])[tp]
  print(thread_name, tp, offset)

  # 設(shè)置消費(fèi)的偏移量
  consumer.seek(tp, offset)

  print u"程序首次運(yùn)行\(zhòng)t線程:", thread_name, u"分區(qū):", partition, u"偏移量:", offset, u"\t開(kāi)始消費(fèi)..."
  num = 0 # 記錄該消費(fèi)者消費(fèi)次數(shù)
  while True:
    msg = consumer.poll(timeout_ms=60000)
    end_offset = consumer.end_offsets([tp])[tp]
    '''可以自己記錄控制消費(fèi)'''
    print u'已保存的偏移量', consumer.committed(tp), u'最新偏移量，', end_offset
    if len(msg) > 0:
      print u"線程:", thread_name, u"分區(qū):", partition, u"最大偏移量:", end_offset, u"有無(wú)數(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的偏移量
        # 手動(dòng)提交偏移量 offsets格式：{TopicPartition:OffsetAndMetadata(offset_num,None)}
        consumer.commit(offsets={tp: (OffsetAndMetadata(end_offset, None))})
        if True == 0:
          # 系統(tǒng)退出？這個(gè)還沒(méi)試
          os.exit()
          '''
          sys.exit()  只能退出該線程，也就是說(shuō)其它兩個(gè)線程正常運(yùn)行，主程序不退出
          '''
      else:
        os.exit()
    else:
      print thread_name, '沒(méi)有數(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")