image.png

1、引入新的Consumer API

0.9.0相比0.8.2，引入了一个新的Consumer API，这个API不再使用high level和low level的基于zookeeper的client；不过仍然支持0.8.0的client。

新的API通过如下方式引入依赖：

   <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.9.0.0</version>
    </dependency>

consumer的offset

kafka0.8.0 的 consumer客户端需要不断与kafka集群的zookeeper交互，以获取最新的offset。而新的consumer的offset则是交给kafka来管理，kafka通过创建专用的topic进行管理不同partition的offset。kafka自己维护了partition的offset，以供同一个partition的不同consumer使用。（图中last commit offset 就是已经确认消费的offset）

image.png

假设现在某一个consumer消费到current position时，未来得及确认已消费就挂掉了，那么下次其他consumer来拉数据时，就从last commit offset开始，重复消费1~6.Consumer的commit。

如果配置了enable.auto.commit为true和auto.commit.interval.ms=xxx，那么就按照这个频率进行commit；

为false时，就需要手动进行commit，可以使用同步方式commitSync，也可以使用 commitAsync 进行异步commit，对于异步确认的话，会返回一个hook，可以利用这个hook进行一定的业务逻辑处理。

consumer通过subscribe方法来订阅它感兴趣的topic，每次订阅之后kafka又有新的consumer加进来的话，那么就要对该topic的position进行重新分配(consumer和partition的比例最好是一个1：1)。

一般而言这个过程是consumer不感兴趣的，因此无需知道；但是如果consumer愿意感知这个事情，那么就可以使用 ConsumerRebalanceListener这个类来进行监听。

另外Consumer可以订阅特殊的partition，实现指定消费partition的功能。适用于一些特殊的场景，比如：消费者所要消费的partition与消费者具有某种联系；或者消费者本身具有高可用性，如果消费者挂掉了，没有必要让kafka来重新分配partition。使用TopicPartition来表示某一个topic的指定partition。

在kafka外部存储offset

允许在kafka外部存储offset，也就是consumer和kafka同时维护一个offset，消费者程序不一定要使用kafka内置的offset存储，而是可以自主选择offset的存储方式。如果能够实现offset和result的原子性保存，将会实现exactly once的事务性保证，要比kafka的offset提交机制所提供的at-least once更加强壮。比如使用外部数据库的事务来保存数据处理结果和offset的一致性，要么共同成功并存储，要么失败回滚。

使用方法：首先将auto.commit提交设置为false，然后使用 ConsumerRecord 来存储offset，需要定位时，使用seek即可。

支持多线程

通过引入wakeupException实现，原理类似于多线程中的InterruptException（通过WakeupException就可以对Consumer进行优雅的控制。而且多个线程公用一个Consumer，Consumer本身非线程安全，因此如果不加外部控制，会导致跑出ConcurrentModificationException。多线程很可能导致非顺序消费数据的问题，但是将消费和业务处理分离，耦合性降低

2、引入了安全管理机制：

• 1. 客户端（producer和consumer）连接broker时，可以使用SSL或者SASL进行验证。
• 2. 验证从broker到zookeeper的连接
• 3. 使用SSL对broker和client之间，broker之间以及使用SSL的工具进行数据编码（这有可能导致性能恶化，取决于CPU和JVM实现）
• 4. 验证客户端的读写操作
• 5. 验证是一个可插拔式的服务，并且支持统一整个验证服务。

SSL或者SASL都是可选择项，如果需要使用，那么就需要进行额外的配置。

3、引入了Kafka Connect：

kafka connect是一个支持Scala的可靠工具。使用它来定义一个数据导入与导出的connector很容易。具有时延低，API操作简单的特征，支持分布式或单机模式。

个人介绍：

高广超：多年一线互联网研发与架构设计经验，擅长设计与落地高可用、高性能、可扩展的互联网架构。

本文首发在 http://www.jianshu.com/u/2766e4cfc391 转载请注明！