如何计算Spring网络流量和千分尺中的Kafka消费者处理时间

Question

我是反应式编程的新手，并在Spring Webflux中编写了一个kafka消费者来消费事件，对其进行处理并处理成功和失败的场景。我想了解如何使用千分尺计算处理每个事件（成功和失败）所花费的时间指标。我知道我们可以使用千分尺的定时器接口来计算这样的处理时间-

Timer timer = Timer.builder("kafka.consumer.time")
                    .tag("eventType", "Event A")
                    .register(meterRegistry);
    timer.record(Duration.ofMillis(System.currentTimeMillis() - inTime));

其中inTime-事件处理开始的时间。

但是我不知道我应该如何在反应式编程场景中使用“inTime ”,因为它是一个事件流，而且它将一次处理多个事件。

 @EventListener(ApplicationReadyEvent.class)
public void consume() {
    kafkaReceiver
            .receive()
            .concatMap(res -> kafkaHelper.process(res)
                    .doOnError(error -> {
                        log.error("Error occurred);
                    }).retryWhen(Retry.backoff(3, Duration.ofSeconds(9)).jitter(0.5))
                    .onErrorResume(error -> {
                        log.error("Retry exhausted);
                        return Mono.empty();
                    })
                    .doOnSuccess(val -> {
                        res.receiverOffset().acknowledge();
                    })).subscribe();

}

请帮忙。提前致谢。

Answer 1

围绕这个问题空间，你可以研究三件事：

< li>Reactor支持千分尺发布，因此您可以命名您的< code>Flux并请求记录如下指标:

receive()
    .name("kafka.events") 
    .metrics() 
    .doOnNext(System.out::println)
    .subscribe();