mq系列kafka01-简介和集群安装

一。kafka简介

    Kafka是一个分布式消息系统，由LinkedLn使用Scala编写，用作LinkedLn的活动流和运营数据处理管道的基础，具有高水平扩展和高吞吐量。它已被多家不同类型的公司作为多种类型的数据管道和消息系统使用。目前越来越多的开源分布式处理系统如Apache Flume、Apache Storm、Spark、ElasticSearch都支持与Kafka集成。
   活动流数据是几乎所有站点在对其网站使用情况做报表时都要用到的数据中最常规的部分。活动数据包括页面访问量（Page View）、被查看内容方面的信息以及搜索情况等内容。这种数据通常的处理方式是先把各种活动以日志的形式写入某种文件，然后周期性地对这些文件进行统计分析。运营数据指的是服务器的性能数据（CPU、IO使用率、请求时间、服务日志等等数据)。运营数据的统计方法种类繁多。近年来，活动和运营数据处理已经成为了网站软件产品特性中一个至关重要的组成部分，这就需要一套稍微更加复杂的基础设施对其提供支持。

设计目标:
  Kafka是一种分布式的、基于发布/订阅的消息系统，主要设计目标如下：
以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间复杂度的访问性能。
  高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条以上消息的传输。
  支持Kafka Server间的消息分区及分布式消费，同时保证每个Partition内的消息顺序传输。
  同时支持离线数据处理和实时数据处理。
  支持在线水平扩展。

消息系统
为何使用消息系统，主要基于以下几点考虑：
解耦。在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。
冗余。有些情况下处理数据的过程会失败，除非数据被持久化，否则将造成数据丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。
扩展性。因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。
灵活性/峰值处理能力。在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。
可恢复性。系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。
顺序保证。在多数使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证了一个Partition内的消息的有序性。
缓冲。在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务最高效率的执行——写入队列的处理会尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。
异步通信。很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

AMQP协议
AMQP即Advanced MessageQueuing Protocol,是一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，不受客户端/中间件不同产品、不同开发语言等条件的限制。AMQP协议模型如下图所示。

消费者（Consumer）：从消息队列中请求消息的客户端应用程序。
生产者（Producer）：向broker发布消息的客户端应用程序。
AMQP服务器端（Broker）：用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

 Kafka架构
Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个Broker。生产者通过网络将消息发送到Kafka集群，集群向消费者提供消息，客户端和服务端通过TCP协议通信，其架构如下图所示。Kafka提供了Java客户端，并且对多种语言都提供了支持。

主题（Topic）：对一组消息的归纳。一个主题类似新闻中的体育、娱乐、教育等分类概念，在实际工程中通常一个业务一个主题。
分区（Partition）：一个Topic中的消息数据按照多个分区组织，分区是Kafka消息队列组织的最小单位，可以看作是一个先入先出（FIFO）的队列。每个分区都由一系列有序的、不可变的消息组成，这些消息被连续的追加到分区中。分区中的每个消息都有一个连续的序列号叫做offset,用来在分区中唯一的标识这个消息。

在一个可配置的时间段内，Kafka集群保留所有发布的消息，不管这些消息有没有被消费。比如，如果消息的保存策略被设置为2天，那么在一个消息被发布的两天时间内，它都是可以被消费的，之后它将被丢弃以释放空间。Kafka的性能是和数据量无关的常量级的，所以保留太多的数据并不是问题。

实际上每个consumer唯一需要维护的数据是消息在日志中的位置，也就是offset。这个offset由consumer来维护：一般情况下随着consumer不断的读取消息，offset的值不断增加，但其实consumer可以以任意的顺序读取消息，比如它可以将offset设置成为一个旧的值来重读之前的消息。

每个分区在Kafka集群的若干服务中都有副本，这些持有副本的服务可以共同处理数据和请求，副本数量是可以配置的。副本使Kafka具备了容错能力：每个分区都有一个服务器作为Leader，零或若干服务器作为Follower，Leader负责处理消息的读和写，Follower则复制Leader。如果Leader宕机了，Follower中的一台则会自动成为新的Leader。集群中的每个服务都会同时扮演两个角色：作为它所持有的一部分分区的Leader，同时作为其他分区的Follower，这样整个集群就会有较好的负载均衡。

Producer将消息发布到它指定的Topic中,并负责决定发布到哪个分区。通常可由负载均衡机制随机选择分区，也可以通过特定的分区函数选择分区。使用的更多的是第二种。

发布消息通常有两种模式：队列模式和发布—订阅模式。队列模式中多个Consumer可以同时从服务端读取消息，每个消息只被其中一个Consumer读到；发布—订阅模式中消息被广播到所有的Consumer中。Consumer可以加入一个Consumer组，共同竞争一个Topic，Topic中的消息将被分发到组中的一个成员中。如果所有的Consumer都在一个组中，这就成为了传统的队列模式。如果所有的Consumer都在不同的组中，这就成为了发布—订阅模式。更常见的是，每个Topic都有若干数量的Consumer组，每个组都是一个逻辑上的订阅者，为了容错和更好的稳定性，每个组由若干Consumer组成。这其实就是一个发布—订阅模式，只不过订阅者是组而不是单个Consumer。

如下所示的Kafka集群由两台机器组成，总共有4个分区和2个Consumer组，A组有2个Consumer而B组有4个。

对比传统消息系统
相比传统的消息系统，Kafka可以很好的保证消息的有序性：
传统的队列在服务器上保存有序的消息，如果多个Consumer同时从这个服务器消费消息，服务器就会以消息存储的顺序向Consumer分发消息。虽然服务器按顺序发布消息，但是消息是被异步的分发到各Consumer上的，所以当消息到达时可能已经失去了原来的顺序，这意味着并发消费将导致顺序错乱。为了避免故障，这样的消息系统通常使用“专用Consumer”的概念，其实就是只允许一个消费者消费消息，当然这就意味着失去了并发性。

在这方面Kafka做的更好。通过分区的概念，Kafka可以在多个Consumer组并发的情况下提供较好的有序性和负载均衡。将每个分区只分发给一个Consumer组，一个分区就只被这个组的一个Consumer消费，这样就可以顺序的消费这个分区的消息。因为有多个分区，依然可以在多个Consumer组之间进行负载均衡。注意Consumer组的数量不能多于分区的数量，也就是有多少分区就允许多少并发消费。

Kafka只能保证一个分区之内消息的有序性，在不同的分区之间是不可以的，这已经可以满足大部分应用的需求。如果需要Topic中所有消息的有序性，那就只能让这个Topic只有一个分区，当然也就只有一个Consumer组消费它。

二。kafka集群安装

   安装Kafka集群之前，需要首先安装JDK和Zookeeper集群  笔记本资源有限 这里就window上有一个zookeeper单机版 就使用这个了（安装集群参考http://blog.csdn.net/liaomin416100569/article/details/71642091）
  模拟环境 

192.168.58.1 zookeeper
192.168.58.149 kafka
192.168.58.150 kafka
192.168.58.151 kafka

安装过程参考官方文档 http://kafka.apache.org/quickstart
下载kafka安装包

wget http://mirror.bit.edu.cn/apache/kafka/1.0.0/kafka_2.11-1.0.0.tgz

1》149安装单机版测试
解压并进入进入解压目录

tar zxvf kafka_2.11-1.0.0.tgz  && cd kafka_2.11-1.0.0

查看目录结构

[root@node2 kafka_2.11-1.0.0]# ll
total 48
drwxr-xr-x 3 root root  4096 Oct 27 09:01 bin
drwxr-xr-x 2 root root  4096 Oct 27 09:01 config
drwxr-xr-x 2 root root  4096 Dec 13 19:02 libs
-rw-r--r-- 1 root root 28824 Oct 27 08:56 LICENSE
-rw-r--r-- 1 root root   336 Oct 27 08:56 NOTICE
drwxr-xr-x 2 root root    43 Oct 27 09:01 site-docs

查看配置文件和执行文件

[root@node2 kafka_2.11-1.0.0]# tree -I 'libs|windows'
.
├── bin
│   ├── connect-distributed.sh
│   ├── connect-standalone.sh
│   ├── kafka-acls.sh
│   ├── kafka-broker-api-versions.sh
│   ├── kafka-configs.sh
│   ├── kafka-console-consumer.sh  消费者接受消息命令
│   ├── kafka-console-producer.sh  生产者发送消息命令
│   ├── kafka-consumer-groups.sh
│   ├── kafka-consumer-perf-test.sh
│   ├── kafka-delete-records.sh
│   ├── kafka-log-dirs.sh
│   ├── kafka-mirror-maker.sh
│   ├── kafka-preferred-replica-election.sh
│   ├── kafka-producer-perf-test.sh
│   ├── kafka-reassign-partitions.sh
│   ├── kafka-replay-log-producer.sh
│   ├── kafka-replica-verification.sh
│   ├── kafka-run-class.sh
│   ├── kafka-server-start.sh 启动kafka服务
│   ├── kafka-server-stop.sh
│   ├── kafka-simple-consumer-shell.sh
│   ├── kafka-streams-application-reset.sh
│   ├── kafka-topics.sh       主题订阅相关命令
│   ├── kafka-verifiable-consumer.sh
│   ├── kafka-verifiable-producer.sh
│   ├── trogdor.sh
│   ├── zookeeper-security-migration.sh
│   ├── zookeeper-server-start.sh #启动内置的zookeeper
│   ├── zookeeper-server-stop.sh
│   └── zookeeper-shell.sh
├── config
│   ├── connect-console-sink.properties
│   ├── connect-console-source.properties
│   ├── connect-distributed.properties
│   ├── connect-file-sink.properties
│   ├── connect-file-source.properties
│   ├── connect-log4j.properties
│   ├── connect-standalone.properties
│   ├── consumer.properties
│   ├── log4j.properties
│   ├── producer.properties
│   ├── server.properties  kafka服务配置文件
│   ├── tools-log4j.properties
│   └── zookeeper.properties  配置内置zookeeper

不使用内置的zookeeper  
修改server.properties 

zookeeper.connect=192.168.58.1:2181

重要的配置解释

broker.id=0 #表示集群的broker的id
log.dirs=/tmp/kafka-logs  #kafka的日志文件位置 最好别放在/tmp目录下
zookeeper.connect=192.168.58.1:2181 #zookeeper的地址

启动kafka

bin/kafka-server-start.sh config/server.properties

创建一个topic主题 1个分区 1个副本

 bin/kafka-topics.sh --create --zookeeper 192.168.58.1:2181 --replication-factor 1 --partitions 1 --topic test

查看所有的topic

[root@node2 kafka_2.11-1.0.0]#  bin/kafka-topics.sh --list --zookeeper 192.168.58.1:2181
test

生产者给topic发送消息

[root@node2 kafka_2.11-1.0.0]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>helloworld
>^C

开启消费者接受消息

[root@node2 kafka_2.11-1.0.0]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
helloworld

2》集群安装
 其他机器 都同上安装kafka (配置参考http://kafka.apache.org/documentation/#adminapi)

58.149配置

broker.id=0
log.dirs=/var/log/kafka-logs
zookeeper.connect=192.168.58.1:2181

58.150配置

broker.id=1
log.dirs=/var/log/kafka-logs
zookeeper.connect=192.168.58.1:2181

58.151配置

broker.id=2
log.dirs=/var/log/kafka-logs
zookeeper.connect=192.168.58.1:2181

分别启动

bin/kafka-server-start.sh config/server.properties

创建一个集群 topic 1个分区 3个副本

bin/kafka-topics.sh --create --zookeeper 192.168.58.1:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

查看该topic到底位于哪个broker服务器

 bin/kafka-topics.sh --describe --zookeeper 192.168.58.1:2181 --topic my-replicated-topic

执行结果

[root@node2 kafka_2.11-1.0.0]#  bin/kafka-topics.sh --describe --zookeeper 192.168.58.1:2181 --topic my-replicated-topic
Topic:my-replicated-topic       PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: my-replicated-topic      Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2

第一行 用于统计分区的结果  PartitionCount表示分区的数量 ReplicationFactor表示副本数量 其他的行就是各个分区的信息 
因为只有一个分区所以只有一行 
 Partition:0 表示分区位于id是0的机器上 也就是149上
 leader就是该分区的主管 负责该分区消息的读和写操作 这里leader是id为1的服务器 也就是150是leader  
 Replicas表示3个副本 副本一般就是（分区的日志）所在机器 这里只有三台 所以分布在 id为0 1 2三台机器上 会列出所有副本机器 不管是leader还是是存活
 Isr表示3个副本 如果其中某些机器挂掉 只会显示存活的机器 
比如我关掉 151的kafka再次尝试查询
 

[root@node2 kafka_2.11-1.0.0]#  bin/kafka-topics.sh --describe --zookeeper 192.168.58.1:2181 --topic my-replicated-topic
Topic:my-replicated-topic       PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: my-replicated-topic      Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 1,0

生产者发送消息 使用任意一台发送即可

 [root@node2 kafka_2.11-1.0.0]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
>hh^He^Hll
>helloworld
>test

使用消费者接受消息

[root@node2 kafka_2.11-1.0.0]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
hll
helloworld
test

容错性
尝试关闭 id：1的leader 尝试生产和接受消息 都可以正常 并且查看选出了新的leader

三。kafka javaapi

 kafka支持以下四种api
The Producer API 用于生产者发布消息.
The Consumer API 用于接受生产者发送的消息并处理.
The Streams API 允许应用扮演一个流处理角色 将生产者输入的数据处理后发送给给输出的消费者
The Connector API 可以使用生产者和消费者连接到第三方的系统或者数据库 用于监听数据的变化

参考http://kafka.apache.org/documentation/#producerapi
添加maven依赖

<dependency>
	    <groupId>org.apache.kafka</groupId>
	    <artifactId>kafka_2.12</artifactId>
	    <version>1.0.0</version>
	</dependency>

1》topic的操作

package kafka;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.security.JaasUtils;

import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZkUtils;
import scala.collection.Iterator;
import scala.collection.Seq;

public class TestTopic {
	/**
	 * 创建topic
	 * bin/kafka-topics.sh --create --zookeeper 192.168.58.1:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic  
	 */
	public static void createTopic(){
		ZkUtils zkUtils = ZkUtils.apply("192.168.58.1:2181", 30000, 30000, JaasUtils.isZkSecurityEnabled());
		AdminUtils.createTopic(zkUtils, "my-topic", 1, 3,  new Properties(), RackAwareMode.Enforced$.MODULE$);
	}
	/**
	 * topic是否存在 bin/kafka-topics.sh --describe --zookeeper 192.168.58.1:2181 --topic my-topic
	 * 以及查詢所有topic bin/kafka-topics.sh --list --zookeeper 192.168.58.1:2181      
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	public static void listTopic() throws InterruptedException, ExecutionException{
		String topicName="my-replicated-topic";
		ZkUtils zkUtils = ZkUtils.apply("192.168.58.1:2181", 30000, 30000, JaasUtils.isZkSecurityEnabled());
		//判斷topic是否存在
		boolean ifExist=AdminUtils.topicExists(zkUtils, topicName);
		System.out.println(ifExist);
		//獲取所有的topic
		Seq<String> allTopics = zkUtils.getAllTopics();
		Iterator<String> iterator = allTopics.iterator();
		while (iterator.hasNext()) {
			String type = (String) iterator.next();
			System.out.println(type);
			
		}
	}
	/**
	 * 刪除topic
	 * bin/kafka-topics.sh --delete --zookeeper 192.168.58.1:2181 --topic my-topic
	 */
	public static void deleteTopic(){
		String topicName="my-topic";
		ZkUtils zkUtils = ZkUtils.apply("192.168.58.1:2181", 30000, 30000, JaasUtils.isZkSecurityEnabled());
		boolean ifExist=AdminUtils.topicExists(zkUtils, topicName);
		if(ifExist){
			AdminUtils.deleteTopic(zkUtils, topicName);
		}
	}
	
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		deleteTopic();
	}
	
}

2》生产者发送消息
  

package kafka;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.security.JaasUtils;

import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZkUtils;
import scala.collection.Iterator;
import scala.collection.Seq;

public class TestProducer {
	public static void main(String[] args) {
		sendMessage();
	}
	/**
	 * 這時官網javaapi的例子
	 * 发送消息
	 */
	public static void sendMessage() {
		//生產者的配置參數類是 ProducerConfig
		 Properties props = new Properties();
		 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.149:9092,192.168.58.150:9092,192.168.58.151:9092");
		 /**
		  * Server完成 producer request 前需要确认的数量。
			acks=0时，producer不会等待确认，直接添加到socket等待发送；
			acks=1时，等待leader写到local log就行；
			acks=all或acks=-1时，等待isr中所有副本确认
			（注意：确认都是 broker 接收到消息放入内存就直接返回确认，不是需要等待数据写入磁盘后才返回确认，这也是kafka快的原因）
		  */
		 props.put(ProducerConfig.ACKS_CONFIG, "all");
		 //發送失敗后的重試次數
		 props.put(ProducerConfig.RETRIES_CONFIG, 0);
		 /**
		  * Producer可以将发往同一个Partition的数据做成一个Produce Request发送请求，即Batch批处理，以减少请求次数，该值即为每次批处理的大小。
			另外每个Request请求包含多个Batch，每个Batch对应一个Partition，且一个Request发送的目的Broker均为这些partition的leader副本。
			若将该值设为0，则不会进行批处理
		  */
		 props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
		 /**
		  * Producer默认会把两次发送时间间隔内收集到的所有Requests进行一次聚合然后再发送，以此提高吞吐量，而linger.ms则更进一步，这个参数为每次发送增加一些delay，以此来聚合更多的Message。
			官网解释翻译：producer会将request传输之间到达的所有records聚合到一个批请求。通常这个值发生在欠负载情况下，record到达速度快于发送。
			但是在某些场景下，client即使在正常负载下也期望减少请求数量。这个设置就是如此，通过人工添加少量时延，而不是立马发送一个record，
			producer会等待所给的时延，以让其他records发送出去，这样就会被聚合在一起。这个类似于TCP的Nagle算法。该设置给了batch的时延上限：
			当我们获得一个partition的batch.size大小的records，就会立即发送出去，而不管该设置；但是如果对于这个partition没有累积到足够的record，
			会linger指定的时间等待更多的records出现。该设置的默认值为0(无时延)。例如，设置linger.ms=5，会减少request发送的数量，
			但是在无负载下会增加5ms的发送时延。
		  */
		 props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
		 /**
		  * Producer可以用来缓存数据的内存大小。该值实际为RecordAccumulator类中的BufferPool，即Producer所管理的最大内存。
			如果数据产生速度大于向broker发送的速度，producer会阻塞max.block.ms，超时则抛出异常
		  */
		 props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
		 // key 序列化方式，类型为class，需实现Serializer interface
		 props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
		 // value 序列化方式，类型为class，需实现Serializer interface
		 props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

		 Producer<String, String> producer = new KafkaProducer<String, String>(props);
		 producer.send(new ProducerRecord<String, String>("my-topic", "1","这是普通消息"));

		 producer.close();
	}

}

3》消费者接受消息

package kafka;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class TestConsumer {
	
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		revMessage();
	}
	/**
	 * 這時官網javaapi的例子
	 * 发送消息
	 */
	public static void revMessage() {
		Properties props = new Properties();
	     props.put("bootstrap.servers", "192.168.58.149:9092,192.168.58.150:9092,192.168.58.151:9092");
	     props.put("group.id", "test");
	     props.put("enable.auto.commit", "false");
	     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
	     consumer.subscribe(Arrays.asList("my-topic"));
	     final int minBatchSize = 200;
	     List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
	     while (true) {
	    	 //poll使用最近的offset去抓取数据 100表示超时时间 后重试
	         ConsumerRecords<String, String> records = consumer.poll(100);
	         for (ConsumerRecord<String, String> record : records) {
	             buffer.add(record);
	             System.out.println(record.key()+"----"+record.value());
	         }
	         if (buffer.size() >= minBatchSize) {
	             consumer.commitSync();
	             buffer.clear();
	         }
	     }
	}
}