使用Resilience4j库具有断路器功能的spring kafka消费者

Question

我正在尝试实现Spring kafka消费者，它需要在处理事件时出现某个异常后暂停（例如：在将事件信息存储到DB时，DB已关闭）。

我们如何在spring boot-2.3.8（spring kafka）中使用Resilience4j断路器方法来处理这种情况

寻找一些消费者暂停和恢复的例子。

@Component
public class CircuitBreakerManager {

    private CircuitBreaker circuitBreaker;

    @Autowired
    private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    public CircuitBreakerManager() {
        CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
                .slidingWindowType(CircuitBreakerConfig.SlidingWindowType.COUNT_BASED)
                .enableAutomaticTransitionFromOpenToHalfOpen()
                .minimumNumberOfCalls(5)
                .permittedNumberOfCallsInHalfOpenState(3)
                .slidingWindowSize(10)
                .failureRateThreshold(50)
                .slowCallRateThreshold(60.0f)
                .slowCallDurationThreshold(Duration.ofSeconds(3))
                .build();
        CircuitBreakerRegistry registry = CircuitBreakerRegistry.of(circuitBreakerConfig);
        this.circuitBreaker = registry.circuitBreaker("serialization_exception");
        this.circuitBreaker.getEventPublisher().onStateTransition(this::onStateChange);
    }

    private void onStateChange(CircuitBreakerOnStateTransitionEvent circuitBreakerEvent) {
        CircuitBreaker.State toState = circuitBreakerEvent.getStateTransition()
                .getToState();
        System.out.println("Change in Circuit Breaker state " + toState);
        switch (toState) {
            case OPEN:
                    kafkaListenerEndpointRegistry.getListenerContainer("my_listener_id").stop();
                break;
            case CLOSED:
                break;
            case HALF_OPEN:
                kafkaListenerEndpointRegistry.getListenerContainer("my_listener_id").start();
                break;
        }
    }


}

在Kafka，listerner只是想捕捉解析错误。如果出现5个以上的解析错误，则需要停止侦听器。但我不确定断路器将如何触发。

@CircuitBreaker(name = RESILIENCE4J_INSTANCE_NAME)
    private Event getParsedEvent(ConsumerRecord consumerRecord) {
        Event event = getEvent(consumerRecord);
        
            if (StringUtils.isEmpty(event)) {
                
                throw new RuntimeException("Serialization Exception occurred");
            }
        }
        return event;
    }

Answer 1

请参阅暂停和恢复侦听器容器。

请注意，只有处理完当前轮询返回的所有记录（或者侦听器抛出异常，只要有默认的错误处理程序），暂停才会生效。