图表设计 网站,哪些网站可以发布免费招聘信息,单页面网站跳出率,个人备案网站可以做电商吗Kafka 位移提交自动提交手动提交Consumer 的消费位移 : 记录 Consumer 下一条消息的消费位移
如 : Consumer 已消费 5 条消息 (位移: 0 - 4) , 此时 Consumer 位移 5 : 指向下一条消息的位移
提交位移 (Committing Offsets) : Consumer 向 Kafka 汇报位移数据
Consumer 能同…
Kafka 位移提交自动提交手动提交Consumer 的消费位移 : 记录 Consumer 下一条消息的消费位移
如 : Consumer 已消费 5 条消息 (位移: 0 - 4) , 此时 Consumer 位移 5 : 指向下一条消息的位移
提交位移 (Committing Offsets) : Consumer 向 Kafka 汇报位移数据
Consumer 能同时消费多个分区的数据Consumer 要维护每个分区提交各自的位移数据当 Consumer 重启后能从之前位移继续消费避免重新消费整个消息
Consumer API 的提交位移的方法 :
从用户分 : 自动提交 , 手动提交从 Consumer 分 : 同步提交 , 异步提交自动提交 : Consumer 在后台提交位移用户无需操作手动提交 : 用户提交位移Consumer 不管
提交位移自动提交配置enable.auto.commit true手动提交同步提交KafkaConsumer.commitSync异步提交KafkaConsumer.commitAsync细化位移提交commitSync(MapTopicPartition, OffsetAndMetadata)commitAsync(MapTopicPartition, OffsetAndMetadata)
自动提交
Consumer 参数 :
enable.auto.commit true : 自动提交位移auto.commit.interval.ms (默认值是 5 秒) : Kafka 每 5 秒自动提交一次位移
自动提交位移
可能出现重复消费例子Consumer 每 5 秒自动提交一次位移。提交位移 3 秒后出现 Rebalance。在 Rebalance 后所有 Consumer 从上一次提交的位移处继续消费但该位移已经是 3 秒前的位移数据在 Rebalance 发生前 3 秒消费的所有数据都会重新消费
设置自动提交位移 :
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, 2000);
props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer);
props.put(value.deserializer, org.apache.kafka.common.serialization.StringDeserializer);KafkaConsumerString, String consumer new KafkaConsumer(props);
consumer.subscribe(Arrays.asList(foo, bar));while (true) {ConsumerRecordsString, String records consumer.poll(100);for (ConsumerRecordString, String record : records) {System.out.printf(offset %d, key %s, value %s%n, record.offset(), record.key(), record.value());}
}手动提交
enable.auto.commit false : 手动提交位移
手动提交位移 :
好处 : 更灵活能把控位移提交的时机和频率缺点 : 用 commitSync() 时Consumer 处于阻塞状态直到 Broker 返回提交结果影响整个应用程序的 TPS
commitSync() :
while (true) {// 返回最新位移。一直等位移提交后才返回 (同步操作)ConsumerRecordsString, String records consumer.poll(Duration.ofSeconds(1));process(records); // 处理消息try {consumer.commitSync();} catch (CommitFailedException e) {handle(e); // 处理提交失败异常}
}commitAsync() :
异步操作会立即返回不会阻塞不影响 Consumer 的 TPS用回调函数 (callback) 实现提交后的逻辑如 : 记录日志或处理异常无法自动失败重试
while (true) {ConsumerRecordsString, String records consumer.poll(Duration.ofSeconds(1));process(records); // 处理消息consumer.commitAsync((offsets, exception) - {if (exception ! null)handle(exception);});
}异步无阻塞式 :
用 commitSync 自动重试避免瞬时错误如 : 网络的瞬时抖动Broker 端 GC异步处理不影响 TPS
// 实现异步无阻塞式的位移管理保证 Consumer 位移的正确性
try {while (true) {ConsumerRecordsString, String records consumer.poll(Duration.ofSeconds(1));process(records); // 处理消息commitAysnc(); // 使用异步提交规避阻塞}
} catch (Exception e) {handle(e); // 处理异常
} finally {try {consumer.commitSync(); // 最后一次提交使用同步阻塞式提交} finally {consumer.close();}
}更精细的位移管理 :
commitSync(MapTopicPartition, OffsetAndMetadata)commitAsync(MapTopicPartition, OffsetAndMetadata)参数 : Map 对象 : 键 TopicPartition (消费的分区)值 OffsetAndMetadata 对象 (位移数据)
// 创建 Map 对象保存 Consumer 消费要提交的分区位移
private MapTopicPartition, OffsetAndMetadata offsets new HashMap();
int count 0;
//...
while (true) {ConsumerRecordsString, String records consumer.poll(Duration.ofSeconds(1));for (ConsumerRecordString, String record: records) {process(record); // 处理消息// 构造要提交的位移值offsets.put(new TopicPartition(record.topic(), record.partition()),new OffsetAndMetadata(record.offset() 1);// 每 100 条消息提交一次位移ifcount % 100 0{consumer.commitAsync(offsets, null); // 回调处理逻辑是 null}count;}
}
