kafka0.9.0 新特性(对比0.8)

  1. 云栖社区>
  2. 博客>
  3. 正文

kafka0.9.0 新特性(对比0.8)

高广超 2017-12-24 21:14:00 浏览630
img_a3684e36984d598b0164ac4a0d3f0bf8.png
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)

img_7c6f4b0a383d5adad13ca6d67309c82b.png
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进行验证。
    1. 验证从broker到zookeeper的连接
    1. 使用SSL对broker和client之间,broker之间以及使用SSL的工具进行数据编码(这有可能导致性能恶化,取决于CPU和JVM实现)
    1. 验证客户端的读写操作
    1. 验证是一个可插拔式的服务,并且支持统一整个验证服务。

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

3、引入了Kafka Connect:

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


个人介绍:

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

本文首发在 高广超的简书博客 转载请注明!

img_31e2e3075b097cabfd9b3643cd9abaa5.png
简书博客
img_3b1610566b09db3358ca5dcb3e015a52.png
头条号