public void startListeners() {
log.debug("get partitions from application");
final List<KafkaPartitionStateKey> partitions = getPartitions();
log.debug("load partition state from hazelcast");
final Map<KafkaPartitionStateKey, KafkaPartitionState> kafkaPartitionStates = kafkaPartitionStateService.loadKafkaPartitionStateMap();
log.debug("check if in sync");
if (areAllPartitionsReady(partitions, kafkaPartitionStates)) {
log.info("all partitions ready, not need to start recovery");
this.messageListenerContainers.forEach(this::startContainer);
return;
}
log.debug("load consumer group offsets from kafka");
consumerGroupOffsets = getConsumerGroupOffsets();
log.debug("create missing partition states");
final List<KafkaPartitionState> updatedPartitionStates = getOrCreatePartitionStates(partitions, kafkaPartitionStates, consumerGroupOffsets);
log.debug("check if all partitions are ready");
if (getNumberOfNotReadyPartitions(updatedPartitionStates) == 0) {
log.info("all partitions ready, no need to start recovery");
this.messageListenerContainers.forEach(this::startContainer);
return;
}
log.info("----- STARTING RECOVERY -----");
this.recoveryListenerContainers.forEach(this::startContainer);
}
2条答案
按热度按时间igsr9ssn1#
在花了几天时间之后,我想出了以下解决办法。
其核心思想是在恢复和正常两种模式下进行同步
在恢复模式中,我只使用数据,但不生成任何数据。
在正常模式下,我消费和产生数据。
在kafka中,我使用了两个属于不同消费群体的侦听器来实现这一点。启动时,所有侦听器都将停止,我决定启用哪种类型的侦听器。一旦所有恢复侦听器的偏移量达到正常侦听器的水印,我就停止恢复侦听器并启动正常侦听器。
下面是我代码的相关部分:
我希望这对某人有用。。。
neekobn82#
这是以前问过的。
使用kafka作为事件存储并不重要,因为问题是微服务重新发送事件。