借助 Spring Kafka,您可以使用pause
and resume
方法取决于 CircuitBreaker 状态转换。我发现最好的方法是将其定义为带有 @Configuration 注释的“supervisor”。还使用了 Resilience4j。
@Configuration
public class CircuitBreakerConsumerConfiguration {
public CircuitBreakerConsumerConfiguration(CircuitBreakerRegistry circuitBreakerRegistry, KafkaManager kafkaManager) {
circuitBreakerRegistry.circuitBreaker("yourCBName").getEventPublisher().onStateTransition(event -> {
switch (event.getStateTransition()) {
case CLOSED_TO_OPEN:
case CLOSED_TO_FORCED_OPEN:
case HALF_OPEN_TO_OPEN:
kafkaManager.pause();
break;
case OPEN_TO_HALF_OPEN:
case HALF_OPEN_TO_CLOSED:
case FORCED_OPEN_TO_CLOSED:
case FORCED_OPEN_TO_HALF_OPEN:
kafkaManager.resume();
break;
default:
throw new IllegalStateException("Unknown transition state: " + event.getStateTransition());
}
});
}
}
这是我与带有注释的 KafkaManager 结合使用的@Component
.
@Component
public class KafkaManager {
private final KafkaListenerEndpointRegistry registry;
public KafkaManager(KafkaListenerEndpointRegistry registry) {
this.registry = registry;
}
public void pause() {
registry.getListenerContainers().forEach(MessageListenerContainer::pause);
}
public void resume() {
registry.getListenerContainers().forEach(MessageListenerContainer::resume);
}
}
此外,我的消费者服务如下所示:
@KafkaListener(topics = "#{'${topic.name}'}", concurrency = "1", id = "CBListener")
public void receive(final ConsumerRecord<String, ReplayData> replayData, Acknowledgment acknowledgment) throws
Exception {
try {
httpClientServiceCB.receiveHandleCircuitBreaker(replayData);
acknowledgement.acknowledge();
} catch (Exception e) {
acknowledgment.nack(1000);
}
}
And the @CircuitBreaker
注解:
@CircuitBreaker(name = "yourCBName")
public void receiveHandleCircuitBreaker(ConsumerRecord<String, ReplayData> replayData) throws
Exception {
try {
String response = restTemplate.getForObject("http://localhost:8081/item", String.class);
} catch (Exception e ) {
// throwing the exception is needed to trigger the Circuit Breaker state change
throw new Exception();
}
}
并且还补充了以下内容application.properties
resilience4j.circuitbreaker.instances.yourCBName.failure-rate-threshold=80
resilience4j.circuitbreaker.instances.yourCBName.sliding-window-type=COUNT_BASED
resilience4j.circuitbreaker.instances.yourCBName.sliding-window-size=5
resilience4j.circuitbreaker.instances.yourCBName.wait-duration-in-open-state=10000
resilience4j.circuitbreaker.instances.yourCBName.automatic-transition-from-open-to-half-open-enabled=true
spring.kafka.consumer.enable.auto.commit = false
spring.kafka.listener.ack-mode = MANUAL_IMMEDIATE
还可以看看https://resilience4j.readme.io/docs/Circuitbreaker https://resilience4j.readme.io/docs/circuitbreaker