Kafka单线程Consumer及参数详解-创新互联
请使用0.9以后的版本：
镜湖ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景，ssl证书未来市场广阔！成为创新互联建站的ssl证书销售渠道，可以享受市场价格4-6折优惠！如果有意向欢迎电话联系或者加微信：18980820575（备注：SSL证书合作）期待与您的合作！示例代码
Properties props = new Properties();
props.put("bootstrap.servers", "kafka01:9092，kafka02: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");
props.put("auto.offset.reset","earliest");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("foo", "bar"));
try{
while (true) {
ConsumerRecords records = consumer.poll(1000);
for (ConsumerRecord record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
}
}finally{
consumer.close();
} 1、只需要配置kafka的server groupid autocommit 序列化 autooffsetreset(其中 bootstrap.server group.id key.deserializer value.deserializer 必须指定);
2、用这些Properties构建consumer对象(KafkaConsumer还有其他构造，可以把序列化传进去);
3、subscribe订阅topic列表（可以用正则订阅Pattern.compile("kafka.*")
使用正则必须指定一个listener subscribe(Pattern pattern, ConsumerRebalanceListener listener)）; 可以重写这个接口来实现 分区变更时的逻辑。如果设置了enable.auto.commit = true 就不用理会这个逻辑。
4、然后循环poll消息(这里的1000是超时设定，如果没有很多数据，也就等一秒);
5、处理消息（打印了offset key value 这里写处理逻辑）。
6、关闭KafkaConsumer（可以传一个timeout值 等待秒数 默认是30）。
参数详解
bootstrap.server(最好用主机名不用ip kafka内部用的主机名 除非自己配置了ip)
deserializer 反序列化consumer从broker端获取的是字节数组，还原回对象类型。
默认有十几种：StringDeserializer LongDeserializer DoubleDeserializer。。
也可以自定义：定义serializer格式 创建自定义deserializer类实现Deserializer 接口 重写逻辑
除了四个必传的 bootstrap.server group.id key.deserializer value.deserializer
还有session.timeout.ms "coordinator检测失败的时间"
是检测consumer挂掉的时间 为了可以及时的rebalance 默认是10秒 可以设置更小的值避免消息延迟。
max.poll.interval.ms "consumer处理逻辑大时间"
处理逻辑比较复杂的时候 可以设置这个值 避免造成不必要的 rebalance ，因为两次poll时间超过了这个参数，kafka认为这个consumer已经跟不上了，会踢出组，而且不能提交offset，就会重复消费。默认是5分钟。
auto.offset.reset "无位移或者位移越界时kafka的应对策略"
所以如果启动了一个group从头消费 成功提交位移后 重启后还是接着消费 这个参数无效
所以3个值的解释是：
earliset 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从最早的位移消费
latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
（注意kafka-0.10.1.X版本之前: auto.offset.reset 的值为smallest,和,largest.(offest保存在zk中) 、
我们这是说的是新版本：kafka-0.10.1.X版本之后: auto.offset.reset 的值更改为:earliest,latest,和none (offest保存在kafka的一个特殊的topic名为:__consumer_offsets里面)）
enable.auto.commit 是否自动提交位移
true 自动提交 false需要用户手动提交 有只处理一次需要的 最近设置为false自己控制。
fetch.max.bytes consumer单次获取大字节数
max.poll.records 单次poll返回的大消息数
默认500条 如果消费很轻量 可以适当提高这个值 增加消费速度。
hearbeat.interval.ms consumer其他组员感知rabalance的时间
该值必须小于 session.timeout.ms 如果检测到 consumer挂掉 也就根本无法感知rabalance了
connections.max.idle.ms 定期关闭连接的时间
默认是9分钟 可以设置为-1 永不关闭
更多实时计算，Kafka等相关技术博文，欢迎关注实时流式计算
另外有需要云服务器可以了解下创新互联scvps.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。
当前题目：Kafka单线程Consumer及参数详解-创新互联
浏览地址：http://mswzjz.cn/article/cchohh.html