Lettuce命令延迟测量(CommandLatency)

Posted 2020-11-14 wei-zw

    Lettuce使用了LatencyUtils进行命令延迟测量,LatencyUtils是一个延迟统计追踪开发包，提供了很多有用的追踪工具.LatencyStats的设计旨在通过简单、嵌入式（drop-in）的延迟行为记录对象，对进程间延迟进行记录和追踪。LatencyStats的功能包括底层追踪和暂停影响纠正、遗漏补偿等。通过可插拔式的暂停监测器与区间估计（interval estimator）结合LatencyStats给出校正后延迟统计直方图。

   测试数据都有什么

下面看看命令延迟统计的结果,可以发现统计了命令的个数,第一个响应的最小延迟,最大延迟以及百分位数的统计,还有响应完成的统计数据

{[local:any -> localhost/127.0.0.1:6379, commandType=GET]=[count=5, timeUnit=MICROSECONDS, 
firstResponse=[min=348, max=518, percentiles={50.0=462, 90.0=518, 95.0=518, 99.0=518, 99.9=518}], 
completion=[min=440, max=8978, percentiles={50.0=544, 90.0=8978, 95.0=8978, 99.0=8978, 99.9=8978}]], 
[local:any -> localhost/127.0.0.1:6379, commandType=SET]=[count=6, timeUnit=MICROSECONDS, 
firstResponse=[min=501, max=15925, percentiles={50.0=581, 90.0=15925, 95.0=15925, 99.0=15925, 99.9=15925}], 
completion=[min=540, max=19267, percentiles={50.0=622, 90.0=19267, 95.0=19267, 99.0=19267, 99.9=19267}]]}

　　

　如何使用命令延迟测量

      在上文中说过,lettuce默认使用的是LatencyUtils作为命令延迟收集器,如果没有更好的选择建议使用默认命令延迟收集器;是不是使用默认命令延迟收集器就什么都不用做了呢?当然不是.下面通过源码走读方式确认一下我们需要做什么?下面是DefaultClientResources 关于命令延迟测量相关源码

        //如果命令延迟收集器为null
        if (builder.commandLatencyCollector == null) {
            //如果默认命令延迟收集器可用
            if (DefaultCommandLatencyCollector.isAvailable()) {
                //如果命令延迟收集器选项不为null,则使用用户自定义都命令延迟收集器选项设置
                if (builder.commandLatencyCollectorOptions != null) {
                    commandLatencyCollector = new DefaultCommandLatencyCollector(builder.commandLatencyCollectorOptions);
                } else {//如果没有设置则使用默认数据
                    commandLatencyCollector = new DefaultCommandLatencyCollector(DefaultCommandLatencyCollectorOptions.create());
                }
            } else {//如果默认命令延迟收集器不可用则将命令延迟收集器选项设置为不可用，并将收集器设置为不可用收集器
                logger.debug("LatencyUtils/HdrUtils are not available, metrics are disabled");
                builder.commandLatencyCollectorOptions = DefaultCommandLatencyCollectorOptions.disabled();
                commandLatencyCollector = DefaultCommandLatencyCollector.disabled();
            }
           //将共享收集器设置为false
            sharedCommandLatencyCollector = false;
        } else {//命令延迟收集器不为null则使用用户指定的命令延迟收集器，并将共享收集器设置为true
            sharedCommandLatencyCollector = true;
            commandLatencyCollector = builder.commandLatencyCollector;
        }
        //命令延迟发射器选项
        commandLatencyPublisherOptions = builder.commandLatencyPublisherOptions;
        //如果命令延迟收集器可用同时命令延迟发射器选项不为null
        if (commandLatencyCollector.isEnabled() && commandLatencyPublisherOptions != null) {
            metricEventPublisher = new DefaultCommandLatencyEventPublisher(eventExecutorGroup, commandLatencyPublisherOptions,
                    eventBus, commandLatencyCollector);
        } else {//如果命令延迟收集器不可用或命令发射选项为null都将测量事件发射器设置为null
            metricEventPublisher = null;
        }

　通过上文源码可以发现使用默认命令延迟测量只需要保证默认命令延迟收集器可用就可以了.那么如何是可用的呢?原来只要在POM中添加LatencyUtils的依赖就可以了

 /**
     * 如果HdrUtils和LatencyUtils在classpath下是有效的就返回true
     */
    public static boolean isAvailable() {
        return LATENCY_UTILS_AVAILABLE && HDR_UTILS_AVAILABLE;
    }

　　

        <dependency>
            <groupId>org.latencyutils</groupId>
            <artifactId>LatencyUtils</artifactId>
            <version>2.0.3</version>
        </dependency>

　　延迟测量数据如何被发送的

通过下面源码可以发现延迟测量数据是通过事件总线发送出去的.同时是按照一个固定的频率发送命令延迟测量数据,这个频率是用户可以配置,如果不配置则默认为10分钟

/**
 * 默认命令延迟事件发射器
 *
 */
public class DefaultCommandLatencyEventPublisher implements MetricEventPublisher {
    //事件处理线程池
    private final EventExecutorGroup eventExecutorGroup;
    //事件发射选项
    private final EventPublisherOptions options;
    //事件总线
    private final EventBus eventBus;
    //命令延迟收集器
    private final CommandLatencyCollector commandLatencyCollector;
    //发射器
    private final Runnable EMITTER = this::emitMetricsEvent;

    private volatile ScheduledFuture<?> scheduledFuture;

    public DefaultCommandLatencyEventPublisher(EventExecutorGroup eventExecutorGroup, EventPublisherOptions options,
            EventBus eventBus, CommandLatencyCollector commandLatencyCollector) {

        this.eventExecutorGroup = eventExecutorGroup;
        this.options = options;
        this.eventBus = eventBus;
        this.commandLatencyCollector = commandLatencyCollector;
        //事件发射间隔不为0
        if (!options.eventEmitInterval().isZero()) {
            //固定间隔发送指标事件
            scheduledFuture = this.eventExecutorGroup.scheduleAtFixedRate(EMITTER, options.eventEmitInterval().toMillis(),
                    options.eventEmitInterval().toMillis(), TimeUnit.MILLISECONDS);
        }
    }

    @Override
    public boolean isEnabled() {
        //指标间隔不为0
        return !options.eventEmitInterval().isZero() && scheduledFuture != null;
    }

    @Override
    public void shutdown() {

        if (scheduledFuture != null) {
            scheduledFuture.cancel(true);
            scheduledFuture = null;
        }
    }

    @Override
    public void emitMetricsEvent() {

        if (!isEnabled() || !commandLatencyCollector.isEnabled()) {
            return;
        }
        //发送命令延迟测试事件
        eventBus.publish(new CommandLatencyEvent(commandLatencyCollector.retrieveMetrics()));
    }

}

　　如何接收到延迟测量数据

我们已经知道延迟测量数据是通过事件总线发送出去的,现在只要订阅事件总线的事件就可以了

client.getResources().eventBus().get().filter(redisEvent -> redisEvent instanceof CommandLatencyEvent)
                .cast(CommandLatencyEvent.class).doOnNext(events::add).subscribe(System.out::println);