kafka基础架构及核心知识_kafka架构

一、kafka介绍以及说明

1.kafka介绍以及名字由来

kafka是一款分布式的基于发布/订阅模式的消息队列，是目前比较主流的消息中间件，Kafka对消息保存时根据Topic（主题）进行归类，发送消息者称为Producer，消息接受者称为Consumer，此外kafka集群有多个kafka实例组成，每个实例(server)称为broker。无论是kafka集群，还是consumer都依赖于zookeeper集群保存一些meta信息，来保证系统可用性,所以安装kafka需要先间搭建zookeeper集群，至于搭建zookeeper集群，请看本人zookeeper介绍那章节。之所以取名kafka，据说是因为开发者非常喜欢奥地利作家卡夫卡，所以以此命名。

 为什么需要zookeeper:Kafka集群中有一个broker会被选举为Controller，负责管理集群broker的上下线，所有topic的分区副本分配和leader选举等工作。Controller的管理工作都是依赖于Zookeeper的。
1

2.kafka数据存储

作为一款消息中间件，很多人误以为写入kafka数据是存储在内存中，但是实际上写入kafka的数据是存储在磁盘中，很多人都认为磁盘很慢，为此，官网专门有一张对此作出了说明(不过听官网语气，大意是人们认为觉得磁盘很慢，但是官网说很快，总结一句话就是：我不要你觉得，我要我觉得)

在官方文档 4.2章持久化的介绍中，官网第一篇说了这样一句话:
这句话的意思是：

 不要害怕文件系统！(文件系统即磁盘) 
 kafka很大程度上是依赖于文件系统来缓存消息。人们普遍认为“磁盘速度很慢”，这使得人们怀疑其(kafka)持久化的架构及性能是否具有竞争力。实际上，磁盘的速度比人期望的更快或者更慢取决于他们(指磁盘)如何被使用。正确设计的磁盘结构通常可以和网络一样快。
1
2

上述的意思就是，磁盘其实并不慢，磁盘的快慢取决于人们如何去使用磁盘，那么，kafka如何在高效的使用磁盘，文档中我标记红框的那部分说了这样一句话：顺序访问磁盘比随机访问内存更快！那么kafka的高效的原因下面就总结出来了。

3.kafka高效的原因

①.顺序写磁盘：Kafka的producer生产数据，要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到600M/s，而随机写只有100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。

②.零拷贝技术：“零拷贝技术”只用将磁盘文件的数据复制到页面缓存中一次，然后将数据从页面缓存直接发送到网络中（发送给不同的订阅者时，都可以使用同一个页面缓存），避免了重复复制操作。如果有10个消费者，传统方式下，数据复制次数为4*10=40次，而使用“零拷贝技术”只需要1+10=11次，一次为从磁盘复制到页面缓存，10次表示10个消费者各自读取一次页面缓存。

③.分区：kafka对每个主题进行分区提高了并发，也提高了效率。
1
2
3
4
5

4.kafka的特点

①类似于消息队列和商业的消息系统，kafka提供对流式数据的发布和订阅
②kafka提供一种持久的容错的方式存储流式数据
③kafka拥有良好的性能，可以及时地处理流式数据
④每条记录由一个键，一个值和一个时间戳组成
1
2
3
4

5.相关单词

producer 生产者[prəˈduːsər]
broker 缓存代理[ˈbroʊkər]
consumers 消费者[kənˈsumərz]
topic 主题[ˈtɑːpɪk]
Interceptor 拦截器[ˌɪntərˈseptər]
Partition 分区[pɑːrˈtɪʃn]

二、kafka集群的安装与部署

安装卡夫卡之前，确保已经安装了zookeeper；进入kafka官网：http://kafka.apache.org/downloads.html 下载

这里以：kafka_2.11-0.11.0.0.tgz为例来搭建集群，例如我有3台机器，机器名分别为hadoop101,hadoop102,hadoop103
1.解压安装包
tar -zxvf kafka_2.11-0.11.0.0.tgz -C /opt
2.修改解压后文件夹名称
mv kafka_2.11-0.11.0.0/ kafka
3.在当前文件夹下创建datas文件夹
mkdir logs
4.修改配置文件
cd conf；
vim server.properties;
配置文件修改如下（以lh01机器为例）：
broker.id=1;
delete.topic.enable=true
log.dirs=/opt/kafka/datas
zookeeper.connect=hadoop101:2181,hadoop102:2181,hadoop103:2181
总共修改4个配置就可以了
5.配置环境变量。将kafka配置到path环境变量下

三、kafka的核心组成

1.Broker

一台kafka服务器就是一个broker。一个集群由多个broker组成，每个broker就是一个kafka的实例。

2.Topic

Topic 就是数据主题，kafka建议根据业务系统将不同的数据存放在不同的topic中！Kafka中的Topics总是多订阅者模式，一个topic可以拥有一个或者多个消费者来订阅它的数据。一个大的Topic可以分布式存储在多个kafka broker中！Topic可以类比为数据库中的库！

3.Interceptor

Interceptor 为拦截器，当生产者向kafka发送数据时，数据会先经过拦截器进行拦截处理，多个拦截器可以组成拦截器链，然后再真正发送数据doSend（）。源代码如下：

@Override
    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
        // intercept the record, which can be potentially modified; this method does not throw exceptions
        ProducerRecord<K, V> interceptedRecord = this.interceptors == null ? record : this.interceptors.onSend(record);
        return doSend(interceptedRecord, callback);
    }
1
2
3
4
5
6

4.Partition

每个topic可以有多个分区，通过分区的设计，topic可以不断进行扩展！即一个Topic的多个分区分布式存储在多个broker；此外通过分区还可以让一个topic被多个consumer进行消费！以达到并行处理！分区可以类比为数据库中的表！kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序。
经过拦截器过滤的代码后，会被进行分区，如果没有指定分区，才会走分区器，所以如果想要自定义分区，不能指定分区，代码如下：

 /**
     * computes partition for given record.
     * if the record has partition returns the value otherwise
     * calls configured partitioner class to compute the partition.
     */
    private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
    //这里尝试获取生产者生产数据的分区号
        Integer partition = record.partition();
        return partition != null ?//如果为null，才会调用分区器
                partition :
                partitioner.partition(
                        record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13

5.Offset

生成者每生产一条数据都会追加到指定分区的log文件中，且存储的记录都是有序的，由于不可随机写入，所以顺序是不变的，这个顺序是通过一个称之为offset的id来唯一标识。
kafka自动维护消费者消费的主题各个分区的offset，前提是消费者消费的分区是由kafka分配的，在启动消费者时，只指定了主题，没有指定分区，kafka会将offset数据保存到一个内置主题为__consumer_offsets的主题中，如果指定了分区，那么kafka将不再自动维护offset。

6.Persistence

Persistence即持久化，Kafka 集群保留所有发布的记录，无论他们是否已被消费，都会通过一个可配置的参数：保留期限来控制。举个例子， 如果保留策略设置为2天，一条记录发布后两天内，可以随时被消费，两天过后这条记录会被清除并释放磁盘空间。
Kafka的性能和数据大小无关，所以长时间存储数据没有什么问题

7.Replication

Replication，即副本，每个分区可能会有多个副本，同一个主题每个分区之间的副本都会选出一个leader，而producer与consumer只与leader之间进行交互，其他follower副本则从leader中同步数据。

8.Producer

消息生产者，就是向kafka broker发消息的客户端。生产者负责将记录分配到topic的指定 partition（分区）中，如果没有指定分区，则都卡夫卡依据分区策略进行分配。

9.Consumer

消息消费者，向kafka broker取消息的客户端。每个消费者都要维护自己读取数据的offset。低版本0.9之前将offset保存在Zookeeper中，0.9及之后保存在Kafka的“__consumer_offsets”主题中
consumer采用pull（拉）模式从broker中读取数据
pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据。针对这一点，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间之后再返回，这段时长即为timeout。

10.Consumer Group

每个消费者都会使用一个消费组名称来进行标识。同一个组中的不同的消费者实例，可以分布在多个进程或多个机器上！
如果所有的消费者实例在同一消费组中，消息记录会负载平衡到每一个消费者实例（单播）。即每个消费者可以同时读取一个topic的不同分区！
如果所有的消费者实例在不同的消费组中，每条消息记录会广播到所有的消费者进程(广播)。
如果需要实现广播，只要每个consumer有一个独立的组就可以了。要实现单播只要所有的consumer在同一个组。
一个topic可以有多个consumer group。topic的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个partion只会把消息发给该CG中的一个consumer。

四、shell客户端操作kafka

1.创建主题
kafka-topics.sh --zookeeper lh02:2181 --create --topic hello1 --partitions 2 --replication-factor 2
创建主题必须指定分区与副本数量，副本数量不能超过当前可用的broker数量；如果只指定了分区数与副本数，由kafka采用负载均衡策略进行对副本自动分配
2.查看主题
①查看所有主题
kafka-topics.sh --zookeeper hadoop102:2181 --list
②查看主题详情
kafka-topics.sh --zookeeper hadoop102:2181 --describe
3.修改主题
kafka-topics.sh --zookeeper hadoop102:2181 --alter --topic first --partitions 6
修改只能修改分区数量以及副本的分配策略，且分区数只能调大，不能调小
4.删除主题
kafka-topics.sh --zookeeper hadoop102:2181 --delete --topic hello1
删除主题分区数据不会马上删除(zookeeper中的元数据会被删除)，只会标记为删除，一段就时间后回收线程会来删除这些被标记的数据。
5.生产消费数据测试
生产者：kafka-console-producer.sh --broker-list hadoop102:9092 --topic hello3
消费者:kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic hello3
消费者默认只会从分区的最后一个数据之后开始消费（默认启动消费者后，只能接受到新生成的数据），消费时，只能保证分区内部有序，不能保证全局有序，如果希望全局有序，那么可以只创建一个分区。
6.查看消费者组
kafka-consumer-groups.sh --bootstrap-server hadoop102:9092 --list

五、kafka工作流程与存储结构

1.存储结构以及文件滚动策略

Kafka采取了分片和索引机制，一个主题可以分为多个区，将每个分区(partition）分为多个片段(segment),每个segment 文件中消息数量不一定相等,每个segment对应两个文件——“.index”文件和“.log”文件，其中index文件为索引文件,log文件为数据文件，segment的log文件大小有配置（log.segment.bytes）决定，默认为1073741824byte=1G,也就是当每个分区生产的消息超过1G之后，就会滚动，产生新的segment 。当然，你也可以指定多长时间滚动一次，不一定要达到1G才滚动，这个配置为log.roll.hours=1或者log.roll.ms=3600000（这个配置需要你自己加上去，kafka给的配置文件中是不存在这个的）

2.工作流程

topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，且每条数据都有自己的offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个offset，以便出错恢复时，从上次的位置继续消费。
那么，消费端是如何知道消费到哪里了？
如果消费的时候，没有指定分区，那么kafka会自动维护offset，当消费端再去消费时，通过offset，根据index索引文件，找到log中对应的位置，然后从下开始继续消费。

六、分区策略、数据的可靠性与一致性

1.分区策略

为什么要分区：
①方便在集群中扩展，一个topic由多个Partition组成，因此整个集群就可以适应任意大小的据
②可以提高并发，因为Partition发送的数据可以以Partition为单位读写

生产数据的分区策略：
首先，生成者发送的数据会被封装成一个ProducerRecord对象，ProducerRecord对象可以指定多个数据：
1.有partition的情况下，直接将指明的值作为partition的值
2.没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值
3.既没有 partition 值又没有 key 值的情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition 值，也就是常说的 round-robin 算法，注意，根据第3条，生产者生产数据时候，如果既没有传key，也没有指定分区，那么第一个数据存放是随机的，之后的数据依次轮询放入各个分区。
消费数据的分区策略：
启动一个消费者组(这个组内有一个或多个消费者)，如果消费时只指定了主题，没有指定分区，系统会自动为当前消费者组内的多个消费者，自动分配分区,分配策略有两种：range，Round_robin。
1.range策略， range以消费者组消费的每个主题为单位，依次列出每个主题当前有多少分区。使用  分区数/消费者数量，如果不能整除，那么排名靠前的消费者会额外多获取一个名额。（如果不能整除，当订阅的分区较多时，排名靠前的消费者压力大！负载不均衡！）
2.Round_robin: 采用轮询的策略分配！轮询采用，先将当前组中所有的消费者订阅的所有的主机和分区汇总,汇总之后（排序），采取轮询的策略分配，但是如果分配的这个分区，当前消费者没有订阅，那么就放弃。
1
2
3
4
5
6
7
8
9
10
11
12
13

如果消费者组内的每个消费者订阅的主题一致，那么轮询相对公平，每个消费者最多消费的分区差1！
订阅的主题和其他消费者差距较大的消费者负载重！

2.数据的可靠性

首先，数据的可靠性根据不同的业务场景有不同的需要，总的来讲就是不丢失数据，如何防止不丢失数据，那么就需要配置ack参数了。ack有3个机制，分别对应如下

0：当ack设置为0时，producer不等待broker的ack，这一操作提供了一个最低的延迟，broker一接收到还没有写入磁盘就已经返回，当broker故障时有可能丢失数据；
1：当ack设置为1时，producer等待broker的ack，partition的leader落盘成功后返回ack，如果在follower同步成功之前leader故障，那么将会丢失数据；
-1（all）：当ack设置为1或all时，这也是kafka的默认策略，producer等待broker的ack，partition的leader和follower全部落盘成功后才返回ack。但是如果在follower同步完成后，broker发送ack之前，leader发生故障，那么会造成数据重复。
1
2
3

那么问题来了，无论ack是0,1还是-1，都会存在数据丢失或者重复消费的问题，那么如何保证数据被精确消费一次呢，根据不同的公司业务逻辑，消费策略也可能不同，但是大体上可以分为三种消费策略：

at least once：最少消费一次，同一条消息可能保存一次或多次,即ack=-1
at most once:最多一次,同一条消息只可能保存一次或0次，即ack=0或ack=1
exactly once:精确一次，同一条消息只能保存一次
1
2
3

现在要面对的就是如何实现 exactly once，即确保消息只被精确的消费一次，但是我们从ack=-1可知，数据至少不会丢失，只有重复的风险，其实解决重复方法有很多种，其中kafka就提供看一种机制：幂等性机制（idempotent），这种机制需要ack=-1使用，，只需将enable.idempotence属性设置为true（其实如果这个参数设置为true了，ack默认就变为-1了），并将retries属性设为Integer.MAX_VALUE。开启上述参数之后，kafka在broker端，对来自每个producer的记录（record）的属性，进行缓存，缓存<producerId,Partition,SequeceNum(id)>数据啊，每次broker将会对这三个参数进行验证，如果相同则代表重复,那么kafka将会不在保存数据。

注意：前提是生产者必须是同一台机器，
1

3.数据的一致性

数据的一致性：每个副本保存的数据都应该是一致的，如果leader宕机，无论哪个副本称为新的leader，消费者消费数据都应该是一致的，即消费者不管从哪个副本消费，消费数据都是相同的。
那么问题就产生了：一个主题每个分区可能多个副本，这些副本有一个leader与多个flower，在kafka中，消费者与生产者只与leader进行通信，其他副本则从leader同步数据，假如leader突然宕机了，而其他副本则都有可能称为leader，那么，哪些副本可以称为leader了，只有在ISR同步队列中副本才有可能称为leader

		ISR：同步队列中的可用副本，Leader维护了一个动态的in-sync replica set (ISR)，意为和leader保持同步的follower集合。当ISR中的follower完成数据的同步之后，leader就会给follower发送ack。如果follower长时间未向leader同步数据，则该follower将被踢出ISR，该时间阈值由replica.lag.time.max.ms参数设定。Leader发生故障之后，就会从ISR中选举新的leader。此队列有leader维护。
		OSR: 不在同步队列的可用副本,如果某个follower迟迟未与leader进行同步，那么leader就会将此副本移动到OSR队列.
1
2

但是如果在follower延迟时间内，leader突然宕机了(假设此时偏移量为20)，但是消费者消费到了17，但是其他副本存储的数据的偏移量为15，当其他副本称为leader之后，消费者从新的leader发现，根本找不到上次消费到17的位置(因为新的leader的最大offset偏移量才15)，此时就产生了问题。
其实kafka只会提供集群中最低offset暴露给消费者，即木桶理论，这里也就引进了两个概念：

	LEO：指的是每个副本最大的offset。
	HW(高水位)：指的是消费者能见到的最大的offset，ISR队列中最小的LEO。(类似于木桶理论中最短的那根木头)，1.1之后改称leader_epoch。
1
2

那么，根据HW这个参数，无论是leader发生故障，还是follower发生故障，都有相应的处理：

	follower故障：follower发生故障后会被临时踢出ISR，待该follower恢复后，follower会读取本地磁盘记录的上次的HW，并将log文件高于HW的部分截取掉，从HW开始向leader进行同步。等该follower的LEO大于等于该Partition的HW，即follower追上leader之后，就可以重新加入ISR了。
	leader故障：leader发生故障之后，会从ISR中选出一个新的leader，之后，为保证多个副本之间的数据一致性，其余的follower会先将各自的log文件高于HW的部分截掉，然后从新的leader同步数据。这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。
1
2

总结:数据的一致性即保证消费者消费的一致性，leader只提供整个所有副本中HW的offset给消费者，消费者也只能消费到offset，这样，无论哪个副本成为新的leader，消费者都可以依据上次消费的位置，继续消费。

七.API实例

1.自定义生产者

package com.lh.test1;

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

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.clients.producer.RecordMetadata;



public class TestInterceptor {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		 Properties props = new Properties();
		 props.put("bootstrap.servers", "hadoop102:9092");
		 props.put("acks", "all");
		 props.put("retries", 0);
		 props.put("batch.size", 16384);
		 props.put("linger.ms", 1);
		 props.put("buffer.memory", 33554432);
		 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		 props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.lh.test1.MyPartition");
		 //设置拦截器
		 List<String> interceptors=new ArrayList<String>();
		 interceptors.add("com.lh.test1.MyCountInterceptor");
		 interceptors.add("com.lh.test1.MyTimeInterceptor");
		 
		 props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptors);
		 Producer<String, String> producer = new KafkaProducer<>(props);
		 for (int i = 0; i < 10; i++) {
			
			 //异步发送
			 //producer.send(new ProducerRecord<String, String>("hello3", Integer.toString(i)));
			 
			 //同步发送
			 RecordMetadata recordMetadata = producer.send(new ProducerRecord<String, String>("hello3", Integer.toString(i),"hahaha"+i)).get();
		     System.out.println(recordMetadata);
		     System.out.println("第"+i+"条数据发送成功！区号:"+recordMetadata.partition()
		     +"本区偏移量："+recordMetadata.offset());
		     Thread.sleep(2000);
		     
		     
		}
		 producer.flush();
		 producer.close();
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

注意：向props放入各种k-v值时候，建议使用ProducerConfig类来获取相应的key，而不是直接使用字符串。

2.自定义消费者

public static void main(String[] args) {
		Properties props = new Properties();
		//建议使用设置key时，使用ConsumerConfig
	     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
	     props.put("group.id", "test");
	     //是否自动提交offset
	     props.put("enable.auto.commit", "false");
	     //提交offset间隔
	     props.put("auto.commit.interval.ms", "1000");
	     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);
	     //如果只指定主题，则可以使用subscribe方法,可以指定多个主题
	     //consumer.subscribe(Arrays.asList("hello3"));
	     //如果既要指定主题，又要指定分区，则使用assign方法
	     List<TopicPartition> topsList=new ArrayList<TopicPartition>();
	     TopicPartition topicPartition=new TopicPartition("hello", 0);
	     topsList.add(topicPartition);   
	     consumer.assign(topsList);
	     //从指定offset读取，如果指定了offset，则提交无论是自动还是手动offset将失效，需要自己维护offset
	     consumer.seek(topicPartition, 0);
	     
	     while (true) {
	    	 //poll从buffer中拉取数据，最多等待10ms
	         ConsumerRecords<String, String> records = consumer.poll(100);
	         for (ConsumerRecord<String, String> record : records) {         
	        	 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
	        	 }

	        
	     }
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

注意：在消费者中，当向props中存入k-v值时，建议使用ConsumerConfig来定义属性值

3.自定义拦截器

拦截器1：

package com.lh.test1;

import java.util.Map;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class MyTimeInterceptor implements ProducerInterceptor<String, String> {

	@Override
	public void configure(Map<String, ?> configs) {
		//读取配置文件
	}

	//拦截处理record
	@Override
	public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
		String value = record.value();
		value=System.currentTimeMillis()+","+value;
		return new ProducerRecord<String, String>(record.topic(),record.key(), value);
	}

	//当收到ack通知时调用
	@Override
	public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
		
	}

	//关闭producer时调用
	@Override
	public void close() {
		// TODO Auto-generated method stub
		
	}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

拦截器2：

package com.lh.test1;

import java.util.Map;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class MyCountInterceptor implements ProducerInterceptor<String, String> {

	private int successCount;
	private int failedCount;
	@Override
	public void configure(Map<String, ?> configs) {
		
	}

	@Override
	public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
		return record;
	}

	@Override
	public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
		if (exception!=null) {
			failedCount++;
		}else {
			successCount++;
		}
	}

	//生成者关闭时输出统计结果
	@Override
	public void close() {
		System.out.println("消息统计:成功："+successCount+"，失败:"+failedCount);
	}
	
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

4.自定义分区器

package com.lh.test1;

import java.util.List;
import java.util.Map;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

public class MyPartition implements Partitioner {

	@Override
	public void configure(Map<String, ?> configs) {
		//获取配置文件
		System.out.println(configs.get("bootstrap.servers"));
	}

	@Override
	public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
		//获取总的分区
		List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
		int numPartitions = partitions.size();
		if (key!=null) {
			int keyInt=Integer.parseInt(key.toString());
			if (keyInt%numPartitions==0) {
				return 0;
			}
			else if (keyInt%numPartitions==1) {
				return 1;
			}
			else {
				return 2;
			}
		}
		else {
			return 0;
		}
	}

	@Override
	public void close() {
		// TODO Auto-generated method stub
		
	}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47