关于Kafka幂等性的原理及实践
Posted WeCoding
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了关于Kafka幂等性的原理及实践相关的知识，希望对你有一定的参考价值。
01 幂等性如此重要
02 哪些因素影响幂等性
先commit，再执行业务逻辑：提交成功，处理失败 。造成丢失
先执行业务逻辑，再commit：提交失败，执行成功。造成重复执行
先执行业务逻辑，再commit：提交成功，异步执行fail。造成丢失
本文将针对以上问题作出讨论
03 Kafka保证发送幂等性
针对以上的问题，kafka在0.11版新增了幂等型producer和事务型producer。前者解决了单会话幂等性等问题，后者解决了多会话幂等性。
单会话幂等性
ProducerStateManager.scala
private def maybeValidateAppend(producerEpoch: Short, firstSeq: Int, offset: Long): Unit = {
validationType match {
case ValidationType.None =>
case ValidationType.EpochOnly =>
checkProducerEpoch(producerEpoch, offset)
case ValidationType.Full =>
checkProducerEpoch(producerEpoch, offset)
checkSequence(producerEpoch, firstSeq, offset)
}
}
private def checkSequence(producerEpoch: Short, appendFirstSeq: Int, offset: Long): Unit = {
if (producerEpoch != updatedEntry.producerEpoch) {
if (appendFirstSeq != 0) {
if (updatedEntry.producerEpoch != RecordBatch.NO_PRODUCER_EPOCH) {
throw new OutOfOrderSequenceException(s"Invalid sequence number for new epoch at offset $offset in " +
s"partition $topicPartition: $producerEpoch (request epoch), $appendFirstSeq (seq. number)")
} else {
throw new UnknownProducerIdException(s"Found no record of producerId=$producerId on the broker at offset $offset" +
s"in partition $topicPartition. It is possible that the last message with the producerId=$producerId has " +
"been removed due to hitting the retention limit.")
}
}
} else {
val currentLastSeq = if (!updatedEntry.isEmpty)
updatedEntry.lastSeq
else if (producerEpoch == currentEntry.producerEpoch)
currentEntry.lastSeq
else
RecordBatch.NO_SEQUENCE
if (currentLastSeq == RecordBatch.NO_SEQUENCE && appendFirstSeq != 0) {
ne throw mew UnknownProducerIdException(s"Local producer state matches expected epoch $producerEpoch " +
s"for producerId=$producerId at offset $offset in partition $topicPartition, but the next expected " +
"sequence number is not known.")
} else if (!inSequence(currentLastSeq, appendFirstSeq)) {
throw new OutOfOrderSequenceException(s"Out of order sequence number for producerId $producerId at " +
s"offset $offset in partition $topicPartition: $appendFirstSeq (incoming seq. number), " +
s"$currentLastSeq (current end sequence number)")
}
}
}
private def inSequence(lastSeq: Int, nextSeq: Int): Boolean = {
nextSeq == lastSeq + 1L || (nextSeq == 0 && lastSeq == Int.MaxValue)
}
引申：Kafka producer 对有序性做了哪些处理
private void insertInSequenceOrder(Deque<ProducerBatch> deque, ProducerBatch batch) {
// When we are requeing and have enabled idempotence, the reenqueued batch must always have a sequence.
if (batch.baseSequence() == RecordBatch.NO_SEQUENCE)
throw new IllegalStateException("Trying to re-enqueue a batch which doesn't have a sequence even " +
"though idempotency is enabled.");
if (transactionManager.nextBatchBySequence(batch.topicPartition) == null)
throw new IllegalStateException("We are re-enqueueing a batch which is not tracked as part of the in flight " +
"requests. batch.topicPartition: " + batch.topicPartition + "; batch.baseSequence: " + batch.baseSequence());
ProducerBatch firstBatchInQueue = deque.peekFirst();
if (firstBatchInQueue != null && firstBatchInQueue.hasSequence() && firstBatchInQueue.baseSequence() < batch.baseSequence()) {
List<ProducerBatch> orderedBatches = new ArrayList<>();
while (deque.peekFirst() != null && deque.peekFirst().hasSequence() && deque.peekFirst().baseSequence() < batch.baseSequence())
orderedBatches.add(deque.pollFirst());
log.debug("Reordered incoming batch with sequence {} for partition {}. It was placed in the queue at " +
"position {}", batch.baseSequence(), batch.topicPartition, orderedBatches.size())
deque.addFirst(batch);
// Now we have to re insert the previously queued batches in the right order.
for (int i = orderedBatches.size() - 1; i >= 0; --i) {
deque.addFirst(orderedBatches.get(i));
}
// At this point, the incoming batch has been queued in the correct place according to its sequence.
} else {
deque.addFirst(batch);
}
}
多会话幂等性
Kafka事务通过隔离机制来实现多会话幂等性
private def maybeValidateAppend(producerEpoch: Short, firstSeq: Int, offset: Long): Unit = {
validationType match {
case ValidationType.None =>
case ValidationType.EpochOnly =>
checkProducerEpoch(producerEpoch, offset)
case ValidationType.Full => //开始事务，执行这个判断
checkProducerEpoch(producerEpoch, offset)
checkSequence(producerEpoch, firstSeq, offset)
}
}
private def checkProducerEpoch(producerEpoch: Short, offset: Long): Unit = {
if (producerEpoch < updatedEntry.producerEpoch) {
throw new ProducerFencedException(s"Producer's epoch at offset $offset is no longer valid in " +
s"partition $topicPartition: $producerEpoch (request epoch), ${updatedEntry.producerEpoch} (current epoch)")
}
}
04 Consumer端幂等性
先commit，再执行业务逻辑：提交成功，处理失败 。造成丢失
先执行业务逻辑，再commit：提交失败，执行成功。造成重复执行
先执行业务逻辑，再commit：提交成功，异步执行fail。造成丢失
对此我们常用的方法时，works取到消息后先执行如下code：
if(cache.contain(msgId)){
// cache中包含msgId，已经处理过
continue;
}else {
lock.lock();
cache.put(msgId,timeout);
commitSync();
lock.unLock();
}
// 后续完成所有操作后，删除cache中的msgId，只要msgId存在cache中，就认为已经处理过。Note：需要给cache设置有消息
如果喜欢我的文章，请长按二维码，关注靳刚同学, 同时您的转发也是对我最大的支持，谢谢！
以上是关于关于Kafka幂等性的原理及实践的主要内容，如果未能解决你的问题，请参考以下文章