kafka consumer 配置详解
1、consumer group 与 topic 订阅
每个consumer 进程都会划归到一个逻辑的consumer group中,逻辑的订阅者是consumer group。所以一条message可以被多个订阅message 所在的topic的每一个consumer group,也就好像是这条message被广播到每个consumer group一样。而每个consumer group中,类似于一个queue(jms中的queue)的概念差不多,即一条消息只会被consumer group中的一个consumer消费。
1.1 consumer 与 partition
其实上面所说的订阅关系还不够明确,其实topic中的partition被分配到某个consumer上,也就是某个consumer订阅了某个partition。 再重复一下:consumer订阅的是partition,而不是message。所以在同一时间点上,订阅到同一个partition的consumer必然属于不同的consumer group。
在官方网站上,给出了这样一张图:
一个kafka cluster中的某个topic,有4个partition。有两个consumer group (a and b)订阅了该topic。 consumer group a有2个partition:p0、p1,consumer group b有4个partition:c3,c4,c5,c6。经过分区分配后,consumer与partition的订阅关系如下:
topic 中的4个partition在consumer group a中的分配情况如下: c1 订阅p0,p3 c2 订阅p1,p2 topic 中的4个partition在consumer group b中的分配情况如下: c3 订阅p0 c4 订阅p3 c5 订阅p1 c6 订阅p2
另外要知道的是,partition分配的工作其实是在consumer leader中完成的。
1.2 consumer 与consumer group
consumer group与consumer的关系是动态维护的:
当一个consumer 进程挂掉 或者是卡住时,该consumer所订阅的partition会被重新分配到该group内的其它的consumer上。当一个consumer加入到一个consumer group中时,同样会从其它的consumer中分配出一个或者多个partition 到这个新加入的consumer。
当启动一个consumer时,会指定它要加入的group,使用的是配置项:group.id。
为了维持consumer 与 consumer group的关系,需要consumer周期性的发送heartbeat到coordinator(协调者,在早期版本,以zookeeper作为协调者。后期版本则以某个broker作为协调者)。当consumer由于某种原因不能发heartbeat到coordinator时,并且时间超过session.timeout.ms时,就会认为该consumer已退出,它所订阅的partition会分配到同一group 内的其它的consumer上。而这个过程,被称为rebalance。
那么现在有这样一个问题:如果一个consumer 进程一直在周期性的发送heartbeat,但是它就是不消费消息,这种状态称为livelock状态。那么在这种状态下,它所订阅的partition不消息是否就一直不能被消费呢?
1.3 coordinator
coordinator 协调者,协调consumer、broker。早期版本中coordinator,使用zookeeper实现,但是这样做,rebalance的负担太重。为了解决scalable的问题,不再使用zookeeper,而是让每个broker来负责一些group的管理,这样consumer就完全不再依赖zookeeper了。
1.3.1 consumer连接到coordinator
从consumer的实现来看,在执行poll或者是join group之前,都要保证已连接到coordinator。连接到coordinator的过程是:
1)连接到最后一次连接的broker(如果是刚启动的consumer,则要根据配置中的borker)。它会响应一个包含coordinator信息(host, port等)的response。
2)连接到coordinator。
1.4 consumer group management
consumer group 管理中,也是需要coordinator的参与。一个consumer要join到一个group中,或者一个consumer退出时,都要进行rebalance。进行rebalance的流程是:
1)会给一个coordinator发起join请求(请求中要包括自己的一些元数据,例如自己感兴趣的topics)
2)coordinator 根据这些consumer的join请求,选择出一个leader,并通知给各个consumer。这里的leader是consumer group 内的leader,是由某个consumer担任,不要与partition的leader混淆。
3)consumer leader 根据这些consumer的metadata,重新为每个consumer member重新分配partition。分配完毕通过coordinator把最新分配情况同步给每个consumer。
4)consumer拿到最新的分配后,继续工作。
2、consumer fetch message
在kafka partition中,每个消息有一个唯一标识,即partition内的offset。每个consumer group中的订阅到某个partition的consumer在从partition中读取数据时,是依次读取的。
上图中,consumer a、b分属于不用的consumer group。consumer b读取到offset =11,consumer a读取到offset=9 。这个值表示consumer group中的某个consumer 在下次读取该partition时会从哪个offset的 message开始读取,即 consumer group a 中的consumer下次会从offset = 9 的message 读取, consumer group b 中的consumer下次会从offset = 11 的message 读取。
这里并没有说是consumer a 下次会从offset = 9 的message读取,原因是consumer a可能会退出group ,然后group a 进行rebalance,即重新分配分区。
2.1 poll 方法
consumer读取partition中的数据是通过调用发起一个fetch请求来执行的。而从kafkaconsumer来看,它有一个poll方法。但是这个poll方法只是可能会发起fetch请求。原因是:consumer每次发起fetch请求时,读取到的数据是有限制的,通过配置项max.partition.fetch.bytes来限制的。而在执行poll方法时,会根据配置项个max.poll.records来限制一次最多pool多少个record。
那么就可能出现这样的情况: 在满足max.partition.fetch.bytes限制的情况下,假如fetch到了100个record,放到本地缓存后,由于max.poll.records限制每次只能poll出15个record。那么kafkaconsumer就需要执行7次才能将这一次通过网络发起的fetch请求所fetch到的这100个record消费完毕。其中前6次是每次pool中15个record,最后一次是poll出10个record。
在consumer中,还有另外一个配置项:max.poll.interval.ms ,它表示最大的poll数据间隔,如果超过这个间隔没有发起pool请求,但heartbeat仍旧在发,就认为该consumer处于 livelock状态。就会将该consumer退出consumer group。所以为了不使consumer 自己被退出,consumer 应该不停的发起poll(timeout)操作。而这个动作 kafkaconsumer client是不会帮我们做的,这就需要自己在程序中不停的调用poll方法了。
2.2 commit offset
当一个consumer因某种原因退出group时,进行重新分配partition后,同一group中的另一个consumer在读取该partition时,怎么能够知道上一个consumer该从哪个offset的message读取呢?也是是如何保证同一个group内的consumer不重复消费消息呢?上面说了一次走网络的fetch请求会拉取到一定量的数据,但是这些数据还没有被消息完毕,consumer就挂掉了,下一次进行数据fetch时,是否会从上次读到的数据开始读取,而导致consumer消费的数据丢失吗?
为了做到这一点,当使用完poll从本地缓存拉取到数据之后,需要client调用commitsync方法(或者commitasync方法)去commit 下一次该去读取 哪一个offset的message。
而这个commit方法会通过走网络的commit请求将offset在coordinator中保留,这样就能够保证下一次读取(不论进行了rebalance)时,既不会重复消费消息,也不会遗漏消息。
对于offset的commit,kafka consumer java client支持两种模式:由kafkaconsumer自动提交,或者是用户通过调用commitsync、commitasync方法的方式完成offset的提交。
自动提交的例子:
properties props = new properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); 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(arrays.aslist("foo", "bar")); while (true) { 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()); }
手动提交的例子:
properties props = new properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); props.put("enable.auto.commit", "false"); 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(arrays.aslist("foo", "bar")); final int minbatchsize = 200; list<consumerrecord<string, string>> buffer = new arraylist<>(); while (true) { consumerrecords<string, string> records = consumer.poll(100); for (consumerrecord<string, string> record : records) { buffer.add(record); } if (buffer.size() >= minbatchsize) { insertintodb(buffer); consumer.commitsync(); buffer.clear(); } }
在手动提交时,需要注意的一点是:要提交的是下一次要读取的offset,例如:
try { while(running) { // 取得消息 consumerrecords<string, string> records = consumer.poll(long.max_value); // 根据分区来遍历数据: for (topicpartition partition : records.partitions()) { list<consumerrecord<string, string>> partitionrecords = records.records(partition); // 数据处理 for (consumerrecord<string, string> record : partitionrecords) { system.out.println(record.offset() + ": " + record.value()); } // 取得当前读取到的最后一条记录的offset long lastoffset = partitionrecords.get(partitionrecords.size() - 1).offset(); // 提交offset,记得要 + 1 consumer.commitsync(collections.singletonmap(partition, new offsetandmetadata(lastoffset + 1))); } } } finally { consumer.close(); }
3、consumer的线程安全性
kafkaproducer是线程安全的,上一节已经了解到。但consumer却没有设计成线程安全的。当用户想要在在多线程环境下使用kafkaconsumer时,需要自己来保证synchronized。如果没有这样的保证,就会抛出concurrentmodificatinexception的。
当你想要关闭consumer或者为也其它的目的想要中断consumer的处理时,可以调用consumer的wakeup方法。这个方法会抛出wakeupexception。
例如:
public class kafkaconsumerrunner implements runnable { private final atomicboolean closed = new atomicboolean(false); private final kafkaconsumer consumer; public void run() { try { consumer.subscribe(arrays.aslist("topic")); while (!closed.get()) { consumerrecords records = consumer.poll(10000); // handle new records } } catch (wakeupexception e) { // ignore exception if closing if (!closed.get()) throw e; } finally { consumer.close(); } } // shutdown hook which can be called from a separate thread public void shutdown() { closed.set(true); consumer.wakeup(); } }
4、consumer configuration
在kafka 0.9+使用java consumer替代了老版本的scala consumer。新版的配置如下:
·bootstrap.servers
在启动consumer时配置的broker地址的。不需要将cluster中所有的broker都配置上,因为启动后会自动的发现cluster所有的broker。
它配置的格式是:host1:port1;host2:port2…
·key.descrializer、value.descrializer
message record 的key, value的反序列化类。
·group.id
用于表示该consumer想要加入到哪个group中。默认值是 “”。
·heartbeat.interval.ms
心跳间隔。心跳是在consumer与coordinator之间进行的。心跳是确定consumer存活,加入或者退出group的有效手段。
这个值必须设置的小于session.timeout.ms,因为:
当consumer由于某种原因不能发heartbeat到coordinator时,并且时间超过session.timeout.ms时,就会认为该consumer已退出,它所订阅的partition会分配到同一group 内的其它的consumer上。
通常设置的值要低于session.timeout.ms的1/3。
默认值是:3000 (3s)
·session.timeout.ms
consumer session 过期时间。这个值必须设置在broker configuration中的group.min.session.timeout.ms 与 group.max.session.timeout.ms之间。
其默认值是:10000 (10 s)
·enable.auto.commit
consumer 在commit offset时有两种模式:自动提交,手动提交。手动提交在前面已经说过。自动提交:是kafka consumer会在后台周期性的去commit。
默认值是true。
·auto.commit.interval.ms
自动提交间隔。范围:[0,integer.max],默认值是 5000 (5 s)
·auto.offset.reset
这个配置项,是告诉kafka broker在发现kafka在没有初始offset,或者当前的offset是一个不存在的值(如果一个record被删除,就肯定不存在了)时,该如何处理。它有4种处理方式:
1) earliest:自动重置到最早的offset。
2) latest:看上去重置到最晚的offset。
3) none:如果边更早的offset也没有的话,就抛出异常给consumer,告诉consumer在整个consumer group中都没有发现有这样的offset。
4) 如果不是上述3种,只抛出异常给consumer。
默认值是latest。
·connections.max.idle.ms
连接空闲超时时间。因为consumer只与broker有连接(coordinator也是一个broker),所以这个配置的是consumer到broker之间的。
默认值是:540000 (9 min)
·fetch.max.wait.ms
fetch请求发给broker后,在broker中可能会被阻塞的(当topic中records的总size小于fetch.min.bytes时),此时这个fetch请求耗时就会比较长。这个配置就是来配置consumer最多等待response多久。
·fetch.min.bytes
当consumer向一个broker发起fetch请求时,broker返回的records的大小最小值。如果broker中数据量不够的话会wait,直到数据大小满足这个条件。
取值范围是:[0, integer.max],默认值是1。
默认值设置为1的目的是:使得consumer的请求能够尽快的返回。
·fetch.max.bytes
一次fetch请求,从一个broker中取得的records最大大小。如果在从topic中第一个非空的partition取消息时,如果取到的第一个record的大小就超过这个配置时,仍然会读取这个record,也就是说在这片情况下,只会返回这一条record。
broker、topic都会对producer发给它的message size做限制。所以在配置这值时,可以参考broker的message.max.bytes 和 topic的max.message.bytes的配置。
取值范围是:[0, integer.max],默认值是:52428800 (5 mb)
·max.partition.fetch.bytes
一次fetch请求,从一个partition中取得的records最大大小。如果在从topic中第一个非空的partition取消息时,如果取到的第一个record的大小就超过这个配置时,仍然会读取这个record,也就是说在这片情况下,只会返回这一条record。
broker、topic都会对producer发给它的message size做限制。所以在配置这值时,可以参考broker的message.max.bytes 和 topic的max.message.bytes的配置。
·max.poll.interval.ms
前面说过要求程序中不间断的调用poll()。如果长时间没有调用poll,且间隔超过这个值时,就会认为这个consumer失败了。
·max.poll.records
consumer每次调用poll()时取到的records的最大数。
·receive.buffer.byte
consumer receiver buffer (so_rcvbuf)的大小。这个值在创建socket连接时会用到。
取值范围是:[-1, integer.max]。默认值是:65536 (64 kb)
如果值设置为-1,则会使用操作系统默认的值。
·request.timeout.ms
请求发起后,并不一定会很快接收到响应信息。这个配置就是来配置请求超时时间的。默认值是:305000 (305 s)
·client.id
consumer进程的标识。如果设置一个人为可读的值,跟踪问题会比较方便。
·interceptor.classes
用户自定义interceptor。
·metadata.max.age.ms
metadata数据的刷新间隔。即便没有任何的partition订阅关系变更也行执行。
范围是:[0, integer.max],默认值是:300000 (5 min)
上一篇: 在struts1中使用FormBean获得请求参数
下一篇: PHP程序运行性能分析