生产者API
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class ProducerMain {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "10.4.121.218:9092");
/**
* acks 控制着认定请求已经完成的标准，all是最慢却最耐用的设置
*/
props.put("acks", "all");
/**
* 如果请求失败了，生产者可以自动重新发送请求的次数
*/
props.put("retries", 0);
/**
* 生产者为每一个partition还没发送出去的记录设置一个缓存区，此配置设置了缓存区的大小，
* 调大参数会导致更大的批处理，但是需要更多的内存，因为会为每一个激活的partition创建
* 这么大的缓存区
*/
props.put("batch.size", 16384);
/**
* 默认情况下，即使缓存区中还有空间，也可以立即发送请求。如果要减少请求次数，可以设置
* linger.ms > 0，这样生产者再发送请求之前会等待该毫秒数，希望有更多的记录被添加到缓
* 存区以填满这个批次的缓存。但是如果1ms之后，仍然没有记录被添加进来，每个请求就会白
* 白增加1ms的延迟。所以在高负载下，可以用一小部分延迟来换取更少、但更高效的请求。另
* 外，即便linger.ms = 0，在一段时间内到达的记录也是会被同一批次处 理的
*/
props.put("linger.ms", 1);
/**
* 生产者可以使用的缓存空间的总量。当请求的发送速度比请求发送到服务器的速度还快，那么缓
* 存空间很快就会被耗尽，这时其他的请求就会被阻塞。阻塞的时长由max.block.ms决定，如果
* 阻塞的时间超过max.block.ms的值，会抛出TimeoutException
*/
props.put("buffer.memory", 33554432);
/**
* serializer指定了如何将ProducerRecord的键值对对象转换成字节。可以使用
* ByteArraySerializer 或者 StringSerializer
*/
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
/**
* 生产者包含一个缓冲池，它保存还未传送到集群上的记录，以及一个后台I/O线程，负责将记录转
* 换成请求，并将其传输到集群。使用后没能正常关闭生产者会泄露缓冲池中的资源。
*/
Producer<String, String> producer = new KafkaProducer<String, String>(props);
for (int i = 0; i < 100; i++) {
/**
* send()方法是异步的，调用时会将记录添加到待处理记录的缓冲池中，然后立即返回
* 这使得生产者可以对缓冲池中的各个记录进行批处理，从而提高效率
*/
producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i)));
}
producer.close();
}
}
消费者API
自动提交
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class AutomaticOffsetConsumer {
public static void main(String[] args) {
Properties props = new Properties();
/**
* 通过bootstrap.servers指定要连接的brokers地址，这个列表仅用于发现集群中的其他代理，
* 并不一定是集群中所有服务器的详尽地址列表。尽管你可能想要指定多个以防客户端连接的时候
* 存在服务器故障。
*/
props.put("bootstrap.servers", "10.4.121.218:9092,10.4.121.219:9092");
/**
* kafka用consumer group的概念，允许一组进程对消费和处理记录的工作进行划分。这些
* 进程要么运行在同一台机器上，要么运行在高可扩展和容错的分布式集群上。所有group.id相同
* 的消费者实例属于同一个consumer group。
* <p>
* 同组中的consumer可以通过 subscribe API 来动态订阅话题列表。然后，kafka会把被订阅
* 话题中的每一条消息分发给每个consumer group的一个进程。这是通过平衡同组所有成员的分区来
* 实现的，这样每个分区正好能分到组中的一个消费者。如果一个主题有4个分区，一个consumer group
* 有2个进程，那么一个进程就消费两个分区。
* <p>
* consumer group中的成员资格是动态维护的：如果进程失败了，原本分配给它的partitions
* 将重新分给同组中的其他进程。类似的，如果一个新的consumer被添加到consumer group中，
* partitions将会被重新分配。这叫group rebalance。当新的partition被添加到被订阅的topic
* 中或新建与订阅的正则表达式匹配的topic时，也会发生group rebalance。group通过定期的元数
* 据刷新来检测是否有新的partitions，并且将新的partitions分配给组中的成员。
* <p>
* 在概念上，可以将consumer group看作是由多个进程组成的逻辑订阅者。作为一个多订阅用户
* 系统，就可以支持任意多个consumer group，对于不同的consumer group，可以重复消费消息，
* 只要订阅主题就可以了。
*/
props.put("group.id", "test");
/**
* 每隔`auto.commit.interval.ms`，自动提交已消费的`offset`，消费者每次获取新数据时都会先把上
* 一次poll()方法返回的最大偏移量提交上去。
* <p>
* 自动提交的问题是会重复消费数据，如果使用默认的 5s 提交时间间隔，在最近一次提交之后的 3s 发生了
* rebalance，rebalance之后，消费者从最后一次提交的偏移量位置开始读取消息，又回到了 3s 之前的
* offset，所以在这 3s内到达的消息会被重复处理。可以通过修改提交时间间隔来更频繁地提交偏移量，
* 减小可能出现重复消息的时间窗，但这种情况是无法完全避免的。
*/
props.put("enable.auto.commit", "true");
// 设置`auto.commit.interval.ms`来控制提交`offset`的频率
props.put("auto.commit.interval.ms", "1000");
/**
* 当 kafka topic 对应的 partition 没有初始`offset`或如果当前`offset`在服务器上不再存在时，
* 共支持三种消费策略。
* 1. earliest: 当无提交的`offset`时，从分区的最开始消费
* 2. latest(default): 当无提交的`offset`时，从分区下最新生产的的数据开始消费
* 3. none: 只要有一个分区不存在已提交的`offset`，则抛出异常
* <p>
* 但是不论哪种策略，如果有当前`group.id`在`topic`的`partition`下有已提交的`offset`，
* 则从提交的`offset`开始消费
*/
props.put("auto.offset.reset", "earliest");
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<String, String>(props);
// 订阅两个主题 foo bar
consumer.subscribe(Arrays.asList("foo", "bar"));
System.out.println("主题订阅:foo bar");
while (true) {
/**
* 订阅完主题之后，当调用poll方法时，consumer会被自动添加到group中。poll()是用来确保消费者
* 存活的方法，只要你一直调用poll方法，consumer就会一直存在group中，接收分配给这个 consumer 的
* partitions中的消息。consumer 会每隔一段时间就发送一次心跳，如果这个 consumer 崩溃了或者超过
* session.timeout.ms设置的时间没有过发送心跳，那么这个consumer就被认为挂了，分配给这个consumer
* 的 partitions 也会被重新分配。
* <p>
* 也有可能consumer一直发送心跳，但是不调用poll去处理消费信息，这样consumer会永远占用分配给它的
* partitions，为了防止这种情况，kafka提供了一种存活检测机制， max.poll.interval.ms。即如果你
* 调用poll()的时间间隔超过了这个值，客户端就会主动将consumer移除出group，然后重新分配partitions。
* 这个机制确保了只有活跃的consumer才能够提交offset，所以为了consumer能保留在group中，你需要一直
* 调用poll方法。
* <p>
* 有两种方法控制poll轮询
* 1. max.poll.interval.ms：提高调用poll的最大间隔时间，这样就可以给consumer更多时间去处理从
* poll中获取的这一批记录。这样做的缺点是会延迟组的rebalance，因为consumer只有在调用poll的时候
* 才被添加到group，然后rebalance。你可以通过这个参数来控制rebalance的频率，但是当实际调用poll
* 的频率达不到这个参数的要求，就会被移出group，这样就会使得处理速度会很慢。
* 2. max.poll.records：调用一次poll最多返回的记录数。这样就能更容易的去预测一次poll间隔时间内
* 要处理多少条记录。调小这个参数值，就可以减少间隔时间，从而减少rebalance的影响。
* <p>
* 对于批次消息处理时间不确定的情况，这些选项都不够。推荐方法是将消息处理移动到另一个线程(异步)，
* 这样消费者就能在处理器处理记录的同时调用poll方法。但是这样就需要禁用自动提交，在线程处理完数据
* 后手动提交记录的offset。需要注意的是提交的offset不能领先实际的position。另外还需要暂停分区，
* 以便在线程处理完之前的记录之后再接受从poll返回的新纪录。
*/
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());
}
}
}
手动提交
同步提交
// 把`auto.commit.offset`设为 false，让应用程序决定何时提交偏移量
props.put("auto.commit.offset", false);
try{
while(true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for(ConsumerRecord<String, String> record : records) {
// 假设把记录内容打印出来就算处理完毕
System.out.println("value = " + record.value() + ", topic = " + record.topic() +
", partition = " + record.partition() + ", offset = " + record.offset());
}
try{
// 只要没有发生不可恢复的错误，commitSync() 方法会一直尝试直至提交成功
// 如果提交失败，我们也只能把异常记录到错误日志里
consumer.commitSync();
}catch(CommitFailedException e) {
System.err.println("commit failed!" + e.getMessage());
}
}
}finally {
consumer.close();
}
异步提交
手动提交有一个不足之处，在 broker 对提交请求作出回应之前，应用程序会一直阻塞，这样会限制应用程序的吞吐量。我们可以通过降低提交频率来提升吞吐量，但如果发生了再均衡，会增加重复消息的数量。
这个时候可以使用异步提交，只管发送提交请求，无需等待 broker 的响应。
// 把auto.commit.offset设为false，让应用程序决定何时提交偏移量
props.put("auto.commit.offset", false);
try{
while(true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for(ConsumerRecord<String, String> record : records) {
System.out.println("value = " + record.value() + ", topic = " + record.topic() +
", partition = " + record.partition() + ", offset = " + record.offset());
}
/**
* 提交最后一个偏移量，然后继续做其他事情。
* <p>
* 在成功提交或碰到无法恢复的错误之前，commitSync()会一直重试，但是commitAsync()不会，
* 它之所以不进行重试，是因为在它收到服务器响应的时候，可能有一个更大的偏移量已经提交成功。
* <p>
* 这也是commitAsync()不好的一个地方。假设我们发出一个请求用于提交偏移量2000，这个时候发
* 生了短暂的通信问题，服务器收不到请求，自然也不会作出任何响应。与此同时，我们处理了另外一
* 批消息，并成功提交了偏移量3000。如果commitAsync()重新尝试提交偏移量2000，它有可能在偏
* 移量3000之后提交成功，这个时候如果发生 rebalance，就会出现重复消费的问题。
*/
consumer.commitAsync();
/*
* commitAsync()也支持回调，在broker作出响应时会执行回调，比如可以在回调中重试失败的提交，
* 以下为思路：
* <p>
* 使用一个单调递增的序列号来维护异步提交的顺序。在每次提交偏移量之后或在回调里提交偏移量时
* 递增序列号。在进行重试前，先检查回调的序列号和即将提交的偏移量是否相等，如果相等，说明没
* 有新的提交，那么可以安全地进行重试。如果序列号比较大，说明有一个新的提交已经发送出去了，
* 应该停止重试。
*/
consumer.commitAsync(new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
if(offsets != null) {
System.out.println("commit offset successful!");
}
if(exception != null) {
System.out.println("commit offset fail!" + exception.getMessage());
}
}
});
}
}finally {
consumer.close();
}
同步和异步组合提交
一般情况下，针对偶尔出现的提交失败，不进行重试不会有太大问题，因为如果提交失败是因为临时问题导致的，那么后续的提交总会有成功的。但如果这是发生在关闭消费者或再均衡前的最后一次提交，就要确保能够提交成功。
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "
+ record.partition() + ", offset = " + record.offset());
}
// 如果一切正常，我们使用 commitAsync() 方法来提交
// 这样速度更快，而且即使这次提交失败，下一次提交很可能会成功
consumer.commitAsync();
}
}catch (Exception e) {
e.printStackTrace();
}finally {
try {
// 使用 commitSync() 方法会一直重试，直到提交成功或发生无法恢复的错误
// 确保关闭消费者之前成功提交了偏移量
consumer.commitSync();
}finally {
consumer.close();
}
}
提交特定的偏移量
不管是自动提交还是使用commitAsync()或者commitSync()来提交偏移量，提交的都是 poll() 方法返回的那批数据的最大偏移量，想要自定义在什么时候提交偏移量可以这么做：
Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
int count = 0;
......
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "
+ record.partition() + ", offset = " + record.offset());
currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset() + 1, "no metadata"));
if (count % 1000 == 0) {
// 这里调用的是 commitAsync()，不过调用 commitSync() 也是完全可以的
// 当然，在提交特定偏移量时，仍然要处理可能发生的错误
consumer.commitAsync(currentOffsets, null);
}
count++;
}
}
}finally {
consumer.close();
}
ReBalance Listener
消费者在退出或者进行分区再均衡之前，应该做一些正确的事情：
- 提交最后一个已处理记录的偏移量(必须做)
- 根据之前处理数据的业务不同，你可能还需要关闭数据库连接池、清空缓存等
程序如何能得知集群要进行”分区再均衡”了？消费者 API 提供了再均衡监听器，以下程序可以做到 kafka 消费数据的 Exactly Once 语义：
package com.bonc.rdpe.kafka110.consumer;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
/**
* @Title RebalanceListenerConsumer.java
* @Description 再均衡监听器
* @Author YangYunhe
* @Date 2018-06-27 17:35:05
*/
public class RebalanceListenerConsumer {
public static void main(String[] args) {
Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.42.89:9092,192.168.42.89:9093,192.168.42.89:9094");
// 把auto.commit.offset设为false，让应用程序决定何时提交偏移量
props.put("auto.commit.offset", false);
props.put("group.id", "dev3-yangyunhe-group001");
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(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {
/*
* 再均衡开始之前和消费者停止读取消息之后被调用
* 如果在这里提交偏移量，下一个接管分区的消费者就知道该从哪里开始读取了
*/
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// 如果发生再均衡，我们要在即将失去分区所有权时提交偏移量
// 要注意，提交的是最近处理过的偏移量，而不是批次中还在处理的最后一个偏移量
System.out.println("Lost partitions in rebalance. Committing current offsets:" + currentOffsets);
consumer.commitSync(currentOffsets);
}
/*
* 在重新分配分区之后和新的消费者开始读取消息之前被调用
*/
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
long committedOffset = -1;
for(TopicPartition topicPartition : partitions) {
// 获取该分区已经消费的偏移量
committedOffset = consumer.committed(topicPartition).offset();
// 重置偏移量到上一次提交的偏移量的下一个位置处开始消费
consumer.seek(topicPartition, committedOffset + 1);
}
}
});
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "
+ record.partition() + ", offset = " + record.offset());
currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset() + 1, "no metadata"));
}
consumer.commitAsync(currentOffsets, null);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try{
consumer.commitSync(currentOffsets);
} catch (Exception e) {
e.printStackTrace();
} finally {
consumer.close();
System.out.println("Closed consumer successfully!");
}
}
}
}
当然你也可以选择再均衡后从头开始消费，与props.put("auto.offset.reset", "earliest");是等效的。
consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
System.out.println("starting partitions rebalance...");
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
consumer.seekToBeginning(partitions);
}
});
设置从最新消息开始消费，与props.put("auto.offset.reset", "latest");等效。
consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
System.out.println("starting partitions rebalance...");
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
consumer.seekToEnd(partitions);
}
});
涉及到数据库的 Exactly Once 语义的实现思路
当处理 Kafka 中的数据涉及到数据库时，那么即使每处理一条数据提交一次偏移量，也可以造成数据重复处理或者丢失数据，看以下为伪代码：
Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
......
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset() + 1);
// 处理数据
processRecord(record);
// 把数据存储到数据库中
storeRecordInDB(record);
// 提交偏移量
consumer.commitAsync(currentOffsets);
}
}
假设把数据存储到数据库后，没有来得及提交偏移量程序就因某种原因挂掉了，那么程序再次启动后就会重复处理数据，数据库中会有重复的数据。
如果把存储到数据库和提交偏移量在一个原子操作里完成，就可以避免这样的问题，但数据存到数据库，偏移量保存到kafka是无法实现原子操作的，而如果把数据存储到数据库中，偏移量也存储到数据库中，这样就可以利用数据库的事务来把这两个操作设为一个原子操作，同时结合再均衡监听器就可以实现 Exactly Once 语义，以下为伪代码：
consumer.subscribe(Collections<String> topics, new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// 发生分区再均衡之前，提交事务
commitDBTransaction();
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
// 再均衡之后，从数据库获得消费偏移量
for(TopicPartition topicPartition : partitions) {
consumer.seek(topicPartition, getOffsetFromDB(topicPartition));
}
}
});
/**
* 消费之前调用一次 poll()，让消费者加入到消费组中，并获取分配的分区
* 然后马上调用 seek() 方法定位分区的偏移量
* seek() 设置消费偏移量，设置的偏移量是从数据库读出来的，说明本次设置的偏移量已经被处理过
* 下一次调用 poll() 就会在本次设置的偏移量上加1，开始处理没有处理过的数据
* 如果seek()发生错误，比如偏移量不存在，则会抛出异常
*/
consumer.poll(0);
for(TopicPartition topicPartition : consumer.assignment()) {
consumer.seek(topicPartition, getOffsetFromDB(topicPartition));
}
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
// 处理数据
processRecord(record);
// 把数据存储到数据库中
storeRecordInDB(record);
// 把偏移量存储到数据库中
storeOffsetInDB(record.topic(), record.partition(), record.offset());
}
// 以上3步为一个事务，提交事务，这里在每个批次末尾提交一次事务，是为了提高性能
commitDBTransaction();
}
把偏移量和记录保存到用一个外部系统来实现 Exactly Once 有很多方法，但核心思想都是：结合 ConsumerRebalanceListener 和 seek() 方法来确保能够及时保存偏移量，并保证消费者总是能够从正确的位置开始读取消息。
