下文介绍如何使用Java从Kafka订阅和读取消息，它和从其它消息系统读取消息有点不同，涉及到一些独特的概念。所以我们要先了解这些概念：

1. Kafka消费者概念

1.1 消费者和消费者组

当你只有一个消费者而且生产者发送消息的速率比消费者读取消息的速率要快的时候，处理新消息就会造成延时，显然需要配置多个消费者去读取消息。Kafka的消费者是消费者组的一部分，当多个消费者订阅一个topic并属于同一个消费者组时，该组中的每个消费者都将会接收这个topic不同分区的消息。

下图Topic T1有4个分区，消费者组G1只有1个消费者C1，那么G1订阅T1的时候，C1会从所有分区读取消息。

如果添加一个消费者C2到G1，每个消费者只会从其中2个分区读取消息。例如C1读取分区0和2，C2读取分区1和3，如下图所示：

如果G1有4个消费者，那么每个消费者只会从其中1个分区读取消息。如下图所示：

如果消费者的数量比分区的数量要多，多出的消费者会处于空闲状态而不会从任何分区读取消息。如下图所示：

Kafka的消费者通常会执行一些高延时的操作，例如是写数据到数据库、对数据进行耗时的计算。在这种情况下，单个消费者不可能及时读取新的消息，因此添加消费者是提高读取消息性能的主要方法。注意，正如上述所说的，消费者的数量不能超过分区的数量，否则会造成资源浪费，因为多出的消费者会处于空闲状态。另外，多个应用从同一个topic读取消息的情况也是非常普遍的。事实上，这也是Kafka的主要设计目标之一。与许多传统的消息系统不同，Kafka在不降低性能的情况下仍然能够支持大量的消费者和消费者组。

例如下图所示，添加一个有2个消费者的消费者组G2，那么G2也会和G1一样读取T1的所有消息。

1.2 消费者组和分区再均衡

当分区的所有权从一个消费者变为另外一个消费者称为分区再均衡。当向消费者组添加一个新的消费者时，它将从之前由另外一个消费者消费的分区中读取消息。当一个消费者被停止或者发生故障时，它会被该消费者组移除，原来由它读取消息的分区会被剩余的其中一个消费者消费。当消费者组消费的topic被修改时（例如，管理员添加新的分区），分区再均衡也会被触发。

分区再均衡是非常重要的，因为它为消费者组提供了高可用性和可扩展性（允许我们容易地和安全地添加和移除消费者），但通常它是不希望发生的。在分区再均衡期间，所有消费者会暂停读取消息，因此分区再均衡基本上会造成整个消费者组的短暂停止。另外，当分区被重新分配给另外一个消费者时，该消费者会丢失其当前的状态；如果它正在缓存任何数据，它将需要刷新它的缓存，这会减缓消息的读取性能直到该消费者重新设置它的状态。

消费者维持分配给它们的分区的所有权是通过向作为该组协调器的Kafka broker发送心跳（不同消费者组的协调器可以是不同的broker）。只要消费者定期发送心跳，它就会被认为是正常的。如果一个消费者停止发送心跳超过一定的时间，其session会超时，该组的协调器会认为它故障并触发分区再均衡。当正常停止一个消费者时，该消费者将通知组协调器它正在离开，组协调器将立即触发分区再均衡。

2. 创建消费者

从Kafka读取消息的第一步是创建一个消费者，类似于创建生产者，也必须指定三个属性：bootstrap.servers、key.deserializer和value.deserializer，第一个之前介绍生产者的时候有详细说明，这里不再重复，简单来说就是用于与Kafka集群建立初始连接的主机和端口的列表。第二个和第三个对应生产者的key.serializer和value.serializer，指定用于把byte数组反序列化为Java对象的类名。

另外还有一个属性group.id，但不是严格强制的，它用于指定该消费者所属的消费者组。

下面是创建消费者的代码示例：

import java.util.Properties;
import org.apache.kafka.clients.consumer.KafkaConsumer;

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("group.id", "CountryCounter");
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);

3. 订阅消息

Kafka允许消费者订阅一个或多个topics的消息，只需要调用subscribe()方法，该方法接收一个需要订阅的topic列表：

import java.util.Collections;

consumer.subscribe(Collections.singletonList("customerCountries"));

也可以使用正则表达式匹配多个topic的名字，下面是匹配前缀为test.的topics：

import java.util.regex.Pattern;

consumer.subscribe(Pattern.compile("test.*"));

注意，在创建一个新的topic时，如果消费者订阅的topic正则表达式匹配新的topic，那么分区再均衡会被立即触发，该消费者会开始从新的topic读取消息。

4. Poll循环

消费者API的核心部分是用一个简单的循环不断地轮询服务器读取新的消息。一旦消费者向topics订阅消息，这个poll循环将会处理协调器的所有操作，例如，分区再均衡、读取消息等。从而为开发人员提供了一个简单地从分配的分区读取数据的API。以下是其主要的实现代码：

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            log.debug("topic = %s, partition = %s, offset = %d, customer = %s, country = %sn",
                record.topic(), record.partition(), record.offset(), record.key(), record.value());
            // Handle new records
        }
    }
} finally {
    consumer.close();
}

poll(long timeout)方法返回一个ConsumerRecords<K, V>，每一个数据都包含topic、分区信息、消息偏移量offset、key和消息值value。新的消费者第一次调用poll()方法时，负责查找组协调者GroupCoordinator、加入到消费者组并与分配的分区连接。如果数据再均衡被触发，它也会同时做相应的处理。注意：消费者是非线程安全的，如果要使用多个消费者，最好使用ExecutorService去启动多线程，而且每个线程使用不同的消费者实例。

5. 消费者配置属性

消费者也有很多配置属性，在0.9.0.0及之后版本，Kafka使用了新的Java消费者，替换了原来基于Scala版本的。除了上述必须的三个之外，下面是一些比较重要的属性：

5.1 fetch.min.bytes

该配置设置broker返回数据的最小字节数，默认是1字节。如果一个broker接收到来自消费者的读取消息请求，但新消息的大小小于该配置的值，那么这个broker会等待更多的消息直到接收的新消息大小等于该配置的值。这样的机制会减少消费者和broker的负载，因为可以减少处理来回消息的频率。所以，在broker接收到的消息不多时，可以把该配置的值设大一点，这样可以减少消费者使用的CPU资源，或者在配置了大量消费者时减少broker的负载。此配置和下面的fetch.max.wait.ms一起使用。

5.2 fetch.max.wait.ms

该配置设置broker最长的等待时间，默认是500ms。如果一个broker接收到的新消息大小小于fetch.min.bytes的值，那么这个broker会一直等待直到时间超过配置的500ms。如果想减少延时，可以把该配置的值设小一点。此配置和上面的fetch.min.bytes是一起使用的，例如fetch.max.wait.ms=100，fetch.min.bytes=1048576(1MB)，当broker数据达到1MB或者等待了100ms时，broker都会把数据发送给消费者。

5.3 max.partition.fetch.bytes

该配置设置每个分区返回数据的最大字节数，默认是1048576字节=1MB。如果一个topic有20个分区，配置了5个消费者，那么每个消费者则需要4MB内存，用于存放KafkaConsumer.poll()返回的ConsumerRecord对象。实际上，如果消费者组内有消费者发生故障，由于分区再均衡，每个消费者则需要处理更多的分区消息，那么建议分配更多的内存。该配置值必须大于broker能够接收消息的最大大小message.max.bytes=1000012(0.96MB)，否则消费者将会hang住。

设置该配置的另一个重要考虑因素是消费者处理数据所花费的时间。消费者必须频繁地调用poll()方法以避免session超时导致分区再均衡，如果单次poll()返回的数据非常大，那么消费者需要更长的时间去处理，这意味着它不能及时调用下一次的poll()方法，从而导致session的超时。这个时候可以减小该配置的值或者增加session的超时时间。

5.4 session.timeout.ms

该配置设置session超时时间，默认是10000=10秒。消费者会周期性地发送心跳给broker，间隔为heartbeat.interval.ms=3000=3秒，如果超过session.timeout.ms配置的时间broker都没有收到心跳，那么对应的消费者就会被移除，然后触发分区再均衡。

5.5 auto.offset.reset

该配置设置消费者在开始读取一个没有提交偏移量或该偏移量为非法的分区时如何重置该偏移量（通常是因为消费者下线时间太长，以至于那个偏移量对应的消息已经过时，或者消息已经被删除）。可用的配置为：

• earliest: 自动重置为最旧的offset，也就是消费者会从最开始读取分区的所有消息
• latest(默认): 自动重置为最新的offset，也就是消费者会从最新的消息开始读取
• none: 向消费者抛出异常
• 其余值: 向消费者抛出异常

5.6 enable.auto.commit

该配置设置是否自动周期性提交offset，默认值为true，提交间隔配置为auto.commit.interval.ms，默认为5000=5秒。

5.7 partition.assignment.strategy

该配置设置消费者实例之间分配分区所有权策略的类名，现有三个可用的类：

org.apache.kafka.clients.consumer.RangeAssignor（默认）：对每个topic，按数字顺序排列可用分区，按字典顺序排列消费者。然后将分区数量除以总消费者数量，以确定分配给每个消费者的分区数量。分区是按数字顺序连续分配给消费者，如果分区没有被均匀分配，那么靠前的消费者会多分配一个分区。例如有2个消费者C0和C1，2个topics t0和t1，每个topic都有3个分区，分别为t0p0, t0p1, t0p2, t1p0, t1p1和t1p2。那么分配给C0，C1的分区分别是C0：[t0p0, t0p1, t1p0, t1p1]，C1：[t0p2, t1p2]。

org.apache.kafka.clients.consumer.RoundRobinAssignor：列出所有可用分区和所有可用消费者，然后把分区循环分配给消费者。如果所有消费者订阅的topic是一样的，那么分区会被均匀分配。例如有和上述RangeAssignor一样的topics，分区和消费者，那么分配给C0，C1的分区分别是C0：[t0p0, t0p2, t1p1]，C1：[t0p1, t1p0, t1p2]。当所有消费者订阅的topic不一样时，分配的时候仍然会按照循环的方式，但会跳过那些没有订阅同一topic的消费者，这可能会导致分区分配不平衡。例如有3个消费者C0，C1和C2，3个topics t0，t1和t2，分别有1，2和3个分区。C0订阅t0；C1订阅t0和t1；C2订阅t0，t1和t2。那么分配给C0，C1和C2的分区分别是C0：[t0p0]，C1：[t1p0]和C2：[t1p1, t2p0, t2p1, t2p2]。

org.apache.kafka.clients.consumer.StickyAssignor：这个类的实现有2个目标：

首先，保证分区分配尽可能均匀，意味着：

• 分配给消费者的分区数量最多相差一个，或者
• 比其它消费者少至少2个分区的消费者无法接收订阅分区的消息

其次，当分区再均衡时，保留尽可能多的现有分配。这会有助于节省当分区从一个消费者重新分配给另一个消费者时的资源消耗。一开始，分区会被尽可能均匀地分配给消费者。虽然这可能听起来类似于RoundRobinAssignor，但下面的第二个例子表明不是。在分区再均衡时，它将按照以下方式执行：

• 分配给消费者的分区仍然尽可能均匀地分配，并且
• 分区尽可能与之前分配的消费者保持一致

当然，上面的第一个目标优先于第二个。

例1，假定有3个消费者C0，C1和C2，4个topics t0，t1，t2和t3，每个topic有2个分区，那么分配给C0，C1和C2的分区分别是C0：[t0p0, t1p1, t3p0]，C1：[t0p1, t2p0, t3p1]和C2：[t1p0, t2p1]。如果C1被移除，触发分区再均衡，RoundRobinAssignor分配的策略是C0：[t0p0, t1p0, t2p0, t3p0]，C2：[t0p1, t1p1, t2p1, t3p1]。而StickyAssignor分配的策略是C0：[t0p0, t1p1, t3p0, t2p0]，C2：[t1p0, t2p1, t0p1, t3p1]。

例2，假定有3个消费者C0，C1和C2，3个topics t0，t1和t2，分别有1，2和3个分区。C0订阅t0；C1订阅t0和t1；C2订阅t0，t1和t2。RoundRobinAssignor分配的策略是C0：[t0p0]，C1：[t1p0]和C2：[t1p1, t2p0, t2p1, t2p2]。而StickyAssignor分配的策略是C0：[t0p0]，C1：[t1p0，t1p1]和C2：[t2p0，t2p1，t2p2]。如果C0被移除，触发分区再均衡，RoundRobinAssignor分配的策略是C1：[t0p0， t1p1]，C2：[t1p0，t2p0，t2p1，t2p2]。而StickyAssignor分配的策略是保留5个分区的分配，C1：[t1p0，t1p1，t0p0]，C2：[t2p0，t2p1，t2p2]。

5.8 client.id

用于标识读取消息的客户端，通常用于日志和性能指标以及配额。

5.9 max.poll.records

该配置设置单次调用poll()方法返回消息的最大数量，默认是500。

5.10 receive.buffer.bytes and send.buffer.bytes

• receive.buffer.bytes：读取数据时使用的TCP接收缓冲区（SO_RCVBUF）的大小，默认值为65536字节=64KB。如果设置为-1，则将使用操作系统的默认值。
• send.buffer.bytes：发送数据时使用的TCP发送缓冲区（SO_SNDBUF）的大小，默认值为131072字节=128KB。如果设置为-1，则将使用操作系统的默认值。

6. 提交Commits和Offsets偏移量

每当调用poll()方法时，它都会返回已经写入Kafka但还没有被当前消费者组读取的消息。Kafka的其中一个独特的特性是它不像其它JMS队列那样跟踪消费者的ACK，而是它允许消费者使用Kafka跟踪每个分区的消费偏移量offset。

消费者向Kafka发送一条包含分区偏移量的消息到一个特别的topic：_consumer_offsets用于更新分区偏移量，这样的操作称为提交commit。在发生分区再均衡时，每个消费者有可能会被分配一些新的分区，这些消费者会从每个分区读取到最新提交的offset开始继续读取消息。如果提交的offset小于之前消费者处理的最后一条消息的offset，那么最后处理的offset与提交的offset之间的消息将会被处理两次，如下图所示：

如果提交的offset大于之前消费者处理的最后一条消息的offset，那么最后处理的offset与提交的offset之间的消息将会丢失，如下图所示：

由此可见，管理offset对客户端应用程序有很大影响。KafkaConsumer API提供了多种提交offset的方法：

6.1 自动提交

提交offset最容易的方法是让消费者自动提交。如果使用了默认的enable.auto.commit=true，那么默认每隔5秒钟(auto.commit.interval.ms)，消费者会自动提交客户端调用poll()方法返回的最大offset。

6.2 同步提交

如果设置了auto.commit.offset=false，那么只有在应用程序调用提交方法时才会提交offset。最简单和最可靠的API是commitSync()，这个API会提交poll()方法返回的最新offset，并在提交成功后才返回，如果由于某种原因提交失败则抛出异常。注意确保在处理完当前的所有消息后才调用commitSync()，否则消息有可能会丢失。以下是示例的代码：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        log.debug("topic = %s, partition = %s, offset = %d, customer = %s, country = %sn",
            record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }
    try {
        consumer.commitSync();
    } catch (CommitFailedException e) {
        log.error("commit failed", e);
    }
}

只要没有不能恢复的异常，commitSync()方法会重试提交。

6.3 异步提交

同步提交的一个缺点是应用程序会被一直阻塞直到有返回，这将会降低应用程序的吞吐量。可以通过减少提交的频率来提高吞吐量，但相应会增加由于分区再均衡而导致的消息被重复处理的数量。这个时候可以选用异步提交，调用提交方法后不需要等待broker的返回。

consumer.commitAsync();

该方法的缺点是不会重试提交，原因是当commitAsync()方法从服务器接收返回时，可能在之后已经有一个成功的提交。例如，当提交offset 2000的时候，消费者和broker出现临时的通讯故障，因此broker暂时接收不了请求。同时，提交另外一个offset 3000，这个时候通讯恢复正常，如果offset 3000先被处理成功，offset 2000重试提交的话，会导致重复处理数据。

另外还有一种带callback参数的commitAsync(OffsetCommitCallback callback)，当接收到broker的返回时会被调用，一般用来记录提交的异常信息或者用于计量性能。以下是示例的代码：

consumer.commitAsync(new OffsetCommitCallback() {
    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
        if (e != null)
            log.error("Commit failed for offsets {}", offsets, e);
    }
});

6.4 结合同步和异步提交

通常情况下，偶尔的提交失败不是一个大的问题，因为如果问题是临时的，那么下一次的提交是会成功的。但如果知道是在关闭消费者之前，或在分区再均衡发生之前的最后一次提交，则需要确保提交成功。因此，常见的做法是结合commitAsync()和commitSync()方法，以下是示例的代码（发生分区再均衡的例子会在后面介绍）：

try {
    consumer.commitAsync();
} catch (CommitFailedException e) {
    log.error("commit failed", e);
} finally {
    try {
        // 关闭消费者之前提交
        consumer.commitSync();
    } finally {
        consumer.close();
    }
}

6.5 提交指定的offset

commitSync()和commitAsync()方法是提交当前批次最新的offset，如果想要在中途提交指定的offset，可以使用：

• commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)
• commitSync(Map<TopicPartition, OffsetAndMetadata> offsets)

以下是示例代码：

Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<TopicPartition, OffsetAndMetadata>();
int count = 0;

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %sn",
                record.topic(), record.partition(), record.offset(), record.key(), record.value());
        currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1, "no metadata"));
        if (count % 1000 == 0)
            // 每处理完1000条消息提交一次
            consumer.commitAsync(currentOffsets, null);
        count++;
    }
}

7. 分区再均衡监听器

当消费者添加或移除分区时，消费者提供了允许执行自定义代码的API：

subscribe(Collection<String> topics, ConsumerRebalanceListener listener)

ConsumerRebalanceListener接口有2个方法需要实现：

• public void onPartitionsRevoked(Collection<TopicPartition> partitions)
• public void onPartitionsAssigned(Collection<TopicPartition> partitions)

onPartitionsRevoked方法是在分区再均衡开始前和消费者停止读取消息后被调用，一般用来提交offset；onPartitionsAssigned方法是在分区重新被分配到broker后和消费者开始读取消息前被调用，下面是使用onPartitionsRevoked()方法在丢失分区所有权之前提交offset的例子：

private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<TopicPartition, OffsetAndMetadata>();

private class HandleRebalance implements ConsumerRebalanceListener {
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 后续会有使用onPartitionsAssigned()的例子
    }

    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("Lost partitions in rebalance. Committing current offsets:" + currentOffsets);
        // 提交已经处理的offset
        consumer.commitSync(currentOffsets);
    }
}

try {
    // 指定使用HandleRebalance监听器
    consumer.subscribe(topics, new HandleRebalance());
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %sn",
                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
            currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1, "no metadata"));
        }
        consumer.commitAsync(currentOffsets, null);
    }
} catch (WakeupException e) {
    // ignore, we're closing
} catch (Exception e) {
    log.error("Unexpected error", e);
} finally {
    try {
        consumer.commitSync(currentOffsets);
    } finally {
        consumer.close();
        System.out.println("Closed consumer and we are done");
    }
}

注意，上例在onPartitionsRevoked()方法提交的是已经处理的offset，不是当前批次的offset。

8. 读取指定offsets的消息

poll()方法是从每个分区最后提交的offset开始读取消息，如果想要从开头开始读取消息，可以使用seekToBeginning(TopicPartition tp)方法；如果想要从结尾开始读取新消息，可以使用seekToEnd(TopicPartition tp)方法。

此外，也可以从指定offset开始读取消息，下面是使用onPartitionsAssigned()方法在重新分配分区后指定offset的例子：

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 如果分区被删除，把相关事务保存在数据库
        commitDBTransaction();
    }

    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for (TopicPartition partition : partitions)
            // 重新分配分区后，从数据库获取保存的offset
            consumer.seek(partition, getOffsetFromDB(partition));
    }
}

// 指定使用SaveOffsetsOnRebalance监听器
consumer.subscribe(topics, new SaveOffsetsOnRebalance());
// 调用poll()方法确保此消费者被添加到消费者里面
consumer.poll(0);

// 获取分区信息并对每一个分区调用seek方法指定开始读取消息的offset
for (TopicPartition partition : consumer.assignment())
    consumer.seek(partition, getOffsetFromDB(partition));
        
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        // 处理消息
        processRecord(record);
        // 保存消息
        storRecordInDB(record);
        // 保存offsets在数据库
        storeOffsetInDB(record.topic(), record.partition(), record.offset());
    }
    // 处理完当前批次消息后，提交事务
    commitDBTransaction();
}

9. 退出poll循环

如果要退出poll()方法，不再读取消息，需要在另外一个线程里面调用consumer.wakeup()方法。如果poll()方法是在主线程运行，那么可以通过ShutdownHook来调用wakeup()方法，例如：

Runtime.getRuntime().addShutdownHook(new Thread() {
    public void run() {
        System.out.println("Starting exit...");
        consumer.wakeup();
        try {
            mainThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});

注意，consumer.wakeup()是唯一可以安全地从其它线程调用的消费者方法，调用此方法会导致poll()方法抛出WakeupException并退出，或者如果在线程没有调用poll()方法之前调用wakeup()，那么WakeupException会在下次调用poll()方法时抛出。该异常不需要处理，但在退出线程之前，必须调用consumer.close()方法关闭消费者，用于提交offsets和向组协调器发送一条消费者离开该组的消息。该组协调器会立即触发分区再均衡，而不需要等待session超时，对应的分区会重新分配给该组的另一个消费者。

try {
    // looping until ctrl-c, the shutdown hook will cleanup on exit
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        // 处理消息
        // ...
    }
} catch (WakeupException e) {
    // 不需要处理WakeupException
} finally {
    // 必须调用close()方法
    consumer.close();
    System.out.println("Closed consumer and we are done");
}

10. 反序列化器

根据之前介绍过的，生产者需要序列号器把对象转为byte数组然后发给broker，相反地，消费者需要反序列化器把从broker读取到的byte数组转为对象。和生产者一样，除了默认提供的反序列化器之外还可以实现自定义的反序列化器，例如：

import java.util.Map;
import org.apache.kafka.common.serialization.Deserializer;

public class CustomerDeserializer implements Deserializer<Customer> {

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // nothing to configure
    }

    @Override
    public Customer deserialize(String topic, byte[] data) {
        // TODO 把byte[]转为Customer对象，这里省略
        return null;
    }

    @Override
    public void close() {
        // nothing to close
    }

}

需要注意的是使用的序列号器和反序列化器必须要一致。

11. 单个的消费者

通常情况下都会配置消费者组，每个组会配置多个消费者。但在某些情况下，如果需要简单化，只配置了一个消费者，而它总是需要从topic的所有分区或特定分区读取消息。这种情况则不需要消费者组和分区再均衡。当确切地知道消费者应该读取哪些分区时，则不需要采用订阅模式，可以直接分配分区。下面是一个消费者如何分配一个topic的所有分区并读取消息的例子：

List<TopicPartition> partitions = new ArrayList<TopicPartition>();
// 读取topic的所有分区
List<PartitionInfo> partitionInfos = consumer.partitionsFor("topic");
if (partitionInfos != null) {
    for (PartitionInfo partition : partitionInfos)
        partitions.add(new TopicPartition(partition.topic(), partition.partition()));
    // 分配所有分区
    consumer.assign(partitions);
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %sn",
                record.topic(), record.partition(), record.offset(), record.key(), record.value());
        }
        consumer.commitSync();
    }
}

注意，如果新添加分区到这个topic，消费者是不会知道的，需要重新调用partitionsFor(String topic)方法才能获取。

END O(∩_∩)O