Kafka Producer Consumer

Producer API

org.apache.kafka.clients.producer.KafkaProducer

如果想学习Java工程化、高性能及分布式、深入浅出。微服务、Spring，MyBatis，Netty源码分析的朋友可以加我的Java高级交流：854630135，群里有阿里大牛直播讲解技术，以及Java大型互联网技术的视频免费分享给大家。

1props.put("bootstrap.servers","192.168.1.128:9092");2props.put("acks","all");3props.put("retries",0);4props.put("batch.size",16384);5props.put("linger.ms",1);6props.put("buffer.memory",33554432);7props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");8props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");910Producer producer = new KafkaProducer(props);11for (int i =0; i <10; i++) {12producer.send(new ProducerRecord("foo", Integer.toString(i), Integer.toString(i)), new Callback() {13@Override14public void onCompletion(RecordMetadata recordMetadata, Exception e) {15if (null != e) {16e.printStackTrace();17}else {18System.out.println("callback: "+ recordMetadata.topic() +" "+ recordMetadata.offset());19}20}21});22}23producer.close();

producer由一个缓冲池组成，这个缓冲池中维护着那些还没有被传送到服务器上的记录，而且有一个后台的I/O线程负责将这些记录转换为请求并将其传送到集群上去。

send()方法是异步的。当调用它以后就把记录放到buffer中并立即返回。这就允许生产者批量的发送记录。

acks配置项控制的是完成的标准，即什么样的请求被认为是完成了的。本例中其值设置的是"all"表示客户端会等待直到所有记录完全被提交，这是最慢的一种方式也是持久化最好的一种方式。

如果请求失败了，生产者可以自动重试。因为这里我们设置retries为0，所以它不重试。

生产者对每个分区都维护了一个buffers，其中放的是未被发送的记录。这些buffers的大小是通过batch.size配置项来控制的。

默认情况下，即使一个buffer还有未使用的空间（PS：buffer没满）也会立即发送。如果你想要减少请求的次数，你可以设置linger.ms为一个大于0的数。这个指令将告诉生产者在发送请求之前先等待多少毫秒，以希望能有更多的记录到达好填满buffer。在本例中，我们设置的是1毫秒，表示我们的请求将会延迟1毫秒发送，这样做是为了等待更多的记录到达，1毫秒之后即使buffer没有被填满，请求也会发送。（PS：稍微解释一下这段话，producer调用send()方法只是将记录放到buffer中，然后由一个后台线程将buffer中的记录传送到服务器上。这里所说的请求指的是从buffer到服务器。默认情况下记录被放到buffer以后立即被发送到服务器，为了减少请求服务器的次数，可以通过设置linger.ms，这个配置项表示等多少毫秒以后再发送，这样做是希望每次请求可以发送更多的记录，以此减少请求次数）

如果想学习Java工程化、高性能及分布式、深入浅出。微服务、Spring，MyBatis，Netty源码分析的朋友可以加我的Java高级交流：854630135，群里有阿里大牛直播讲解技术，以及Java大型互联网技术的视频免费分享给大家。

buffer.memory控制的是总的buffer内存数量

key.serializer 和 value.serializer表示怎样将key和value对象转成字节

从kafka 0.11开始，KafkaProducer支持两种模型：the idempotent producer and the transactional producer（幂等producer和事务producer）。幂等producer强调的是至少一次精确的投递。事务producer允许应用程序原子的发送消息到多个分区或者主题。

为了启用幂等性，必须将enable.idempotence这个配置的值设为true。如果你这样设置了，那么retries默认是Integer.MAX_VALUE，并且acks默认是all。为了利用幂等producer的优势，请避免应用程序级别的重新发送。

为了使用事务producer，你必须配置transactional.id。如果transactional.id被设置，幂等性自动被启用。

1Properties props = new Properties();2props.put("bootstrap.servers","192.168.1.128:9092");3props.put("transactional.id","my-transactional-id");45Producer producer = new KafkaProducer(props, new StringSerializer(), new StringSerializer());67producer.initTransactions();89try {10producer.beginTransaction();1112for (int i =11; i <20; i++) {13producer.send(new ProducerRecord("bar", Integer.toString(i), Integer.toString(i)));14}15// This method will flush any unsent records before actually committing the transaction16producer.commitTransaction();17} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {18producer.close();19} catch (KafkaException e) {20// By calling producer.abortTransaction() upon receiving a KafkaException we can ensure21// that any successful writes are marked as aborted, hence keeping the transactional guarantees.22producer.abortTransaction();23}2425producer.close();

如果想学习Java工程化、高性能及分布式、深入浅出。微服务、Spring，MyBatis，Netty源码分析的朋友可以加我的Java高级交流：854630135，群里有阿里大牛直播讲解技术，以及Java大型互联网技术的视频免费分享给大家。

Consumer API

org.apache.kafka.clients.consumer.KafkaConsumer

Offsets and Consumer Position

对于分区中的每条记录，kafka维护一个数值偏移量。这个偏移量是分区中一条记录的唯一标识，同时也是消费者在分区中的位置。例如，一个消费者在分区中的位置是5，表示它已经消费了偏移量从0到4的记录，并且接下来它将消费偏移量为5的记录。相对于消费者用户来说，这里实际上有两个位置的概念。

消费者的position表示下一条将要消费的记录的offset。每次消费者通过调用poll(long)接收消息的时候这个position会自动增加。

committed position表示已经被存储的最后一个偏移量。消费者可以自动的周期性提交offsets，也可以通过调用提交API(e.g. commitSync and commitAsync)手动的提交position。

Consumer Groups and Topic Subscriptions

Kafka用"consumer groups"（消费者组）的概念来允许一组进程分开处理和消费记录。这些处理在同一个机器上进行，也可以在不同的机器上。同一个消费者组中的消费者实例有相同的group.id

组中的每个消费者可以动态设置它们想要订阅的主题列表。Kafka给每个订阅的消费者组都投递一份消息。这归功于消费者组中所有成员之间的均衡分区，以至于每个分区都可以被指定到组中精确的一个消费者。假设一个主题有4个分区，一个组中有2个消费者，那么每个消费者将处理2个分区。

消费者组中的成员是动态维护的：如果一个消费者处理失败了，那么分配给它的分区将会被重新分给组中其它消费者。

在概念上，你可以把一个消费者组想象成一个单个的逻辑订阅者，并且每个逻辑订阅者由多个进程组成。作为一个多订阅系统，Kafka天生就支持对于给定的主题可以有任意数量的消费者组。

Automatic Offset Committing

1Propertiesprops=newProperties();2props.put("bootstrap.servers","192.168.1.128:9092");3props.put("group.id","test");4props.put("enable.auto.commit","true");5props.put("auto.commit.interval.ms","1000");6props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");7props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");8KafkaConsumer consumer =newKafkaConsumer(props);9consumer.subscribe(Arrays.asList("foo","bar"));10while(true) {11ConsumerRecords records = consumer.poll(100);12for(ConsumerRecord record : records) {13System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());14}15}

设置enable.auto.commit意味着自动提交已消费的记录的offset

Manual Offset Control

代替消费者周期性的提交已消费的offsets，用户可以控制什么时候记录被认为是已经消费并提交它们的offsets。

1Properties props =newProperties();2props.put("bootstrap.servers","localhost:9092");3props.put("group.id","test");4props.put("enable.auto.commit","false");5props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");6props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");7KafkaConsumer consumer =newKafkaConsumer<>(props);8consumer.subscribe(Arrays.asList("foo","bar"));9finalintminBatchSize =200;10List>buffer=newArrayList<>();11while(true) {12ConsumerRecords records = consumer.poll(100);13for(ConsumerRecord record : records) {14buffer.add(record);15}16if(buffer.size() >= minBatchSize) {17insertIntoDb(buffer);18consumer.commitSync();19buffer.clear();20}21}

代码演示

服务器端

客户端

如果想学习Java工程化、高性能及分布式、深入浅出。微服务、Spring，MyBatis，Netty源码分析的朋友可以加我的Java高级交流：854630135，群里有阿里大牛直播讲解技术，以及Java大型互联网技术的视频免费分享给大家。