這個 Redis 連線池的新監控方式針不戳~我再加一點佐料

語言: CN / TW / HK

Lettuce 是一個 Redis 連線池,和 Jedis 不一樣的是,Lettuce 是主要基於 Netty 以及 ProjectReactor 實現的非同步連線池。由於基於 ProjectReactor,所以可以直接用於 spring-webflux 的非同步專案,當然,也提供了同步介面。

在我們的微服務專案中,使用了 Spring Boot 以及 Spring Cloud 。並且使用了 spring-data-redis 作為連線 Redis 的庫。並且連線池使用的是 Lettuce。同時,我們線上的 JDK 是 OpenJDK 11 LTS 版本,並且 每個程序都打開了 JFR 記錄 。關於 JFR,可以參考這個系列:[JFR 全解]()

在 Lettuce 6.1 之後,Lettuce 也引入了基於 JFR 的監控事件。參考: events.flight-recorder

1. Redis 連線相關事件:

  • ConnectEvent :當嘗試與 Redis 建立連線之前,就會發出這個事件。
  • ConnectedEvent連線建立的時候會發出的事件 ,包含建立連線的遠端 IP 與埠以及使用的 Redis URI 等資訊,對應 Netty 其實就是 ChannelHandler 中的 channelActive 回撥一開始就會發出的事件。
  • ConnectionActivatedEvent :在完成 Redis 連線一系列初始化操作之後(例如 SSL 握手,傳送 PING 心跳命令等等), 這個連線可以用於執行 Redis 命令時發出的事件
  • ConnectionDeactivatedEvent :在沒有任何正在處理的命令並且 isOpen() 是 false 的情況下, 連線就不是活躍的了 ,準備要被關閉。這個時候就會發出這個事件。
  • DisconnectedEvent連線真正關閉或者重置時 ,會發出這個事件。
  • ReconnectAttemptEvent :Lettuce 中的 Redis 連線會被維護為長連線, 當連線丟失,會自動重連,需要重連的時候 ,會發出這個事件。
  • ReconnectFailedEvent :當重連並且失敗的時候的時候,會發出這個事件。

2. Redis 叢集相關事件:

  • AskRedirectionEvent :針對 Redis slot 處於遷移狀態時會返回 ASK,這時候會發出這個事件。
  • MovedRedirectionEvent :針對 Redis slot 不在當前節點上時會返回 MOVED,這時候會發出這個事件。
  • TopologyRefreshEvent :如果啟用了叢集拓補重新整理的定時任務,在查詢叢集拓補的時候,就會發出這個事件。但是,這個需要在配置中開啟定時檢查叢集拓補的任務,參考 cluster-topology-refresh
  • ClusterTopologyChangedEvent :當 Lettuce 發現 Redis 叢集拓補發生變化的時候,就會發出這個事件。

3. Redis 命令相關事件:

  • CommandLatencyEvent :Lettuce 會統計每個命令的響應時間,並定時發出這個事件。這個也是需要手動配置開啟的,後面會提到如何開啟。
  • CommandStartedEvent開始執行某一指令 的時候會發出這個事件。
  • CommandSucceededEvent指令執行成功 的時候會發出這個事件。
  • CommandFailedEvent指令執行失敗 的時候會發出這個事件。

Lettuce 的監控是基於事件分發與監聽機制的設計,其核心介面是 EventBus :

EventBus.java

public interface EventBus {
    // 獲取 Flux,通過 Flux 訂閱,可以允許多個訂閱者
    Flux<Event> get();
    // 釋出事件
    void publish(Event event);
}

其預設實現為 DefaultEventBus

public class DefaultEventBus implements EventBus {
    private final DirectProcessor<Event> bus;
    private final FluxSink<Event> sink;
    private final Scheduler scheduler;
    private final EventRecorder recorder = EventRecorder.getInstance();

    public DefaultEventBus(Scheduler scheduler) {
        this.bus = DirectProcessor.create();
        this.sink = bus.sink();
        this.scheduler = scheduler;
    }

    @Override
    public Flux<Event> get() {
        //如果消費不過來直接丟棄
        return bus.onBackpressureDrop().publishOn(scheduler);
    }

    @Override
    public void publish(Event event) {
        //呼叫 recorder 記錄
        recorder.record(event);
        //呼叫 recorder 記錄之後,再發布事件
        sink.next(event);
    }
}

在預設實現中,我們發現釋出一個事件首先要呼叫 recorder 記錄,之後再放入 FluxSink 中進行事件釋出。目前 recorder 有實際作用的實現即基於 JFR 的 JfrEventRecorder .檢視原始碼:

JfrEventRecorder

public void record(Event event) {
    LettuceAssert.notNull(event, "Event must not be null");
    //使用 Event 建立對應的 JFR Event,之後直接 commit,即提交這個 JFR 事件到 JVM 的 JFR 記錄中
    jdk.jfr.Event jfrEvent = createEvent(event);
    if (jfrEvent != null) {
        jfrEvent.commit();
    }
}

private jdk.jfr.Event createEvent(Event event) {
    try {
        //獲取構造器,如果構造器是 Object 的構造器,代表沒有找到這個 Event 對應的 JFR Event 的構造器
        Constructor<?> constructor = getEventConstructor(event);
        if (constructor.getDeclaringClass() == Object.class) {
            return null;
        }
        //使用構造器建立 JFR Event
        return (jdk.jfr.Event) constructor.newInstance(event);
    } catch (ReflectiveOperationException e) {
        throw new IllegalStateException(e);
    }
}

//Event 對應的 JFR Event 構造器快取
private final Map<Class<?>, Constructor<?>> constructorMap = new HashMap<>();

private Constructor<?> getEventConstructor(Event event) throws NoSuchMethodException {
    Constructor<?> constructor;
    //簡而言之,就是檢視快取 Map 中是否存在這個 class 對應的 JFR Event 構造器,有則返回,沒有則嘗試發現
    synchronized (constructorMap) {
        constructor = constructorMap.get(event.getClass());
    }
    if (constructor == null) {
    
        //這個發現的方式比較粗暴,直接尋找與當前 Event 的同包路徑下的以 Jfr 開頭,後面跟著當前 Event 名稱的類是否存在
        //如果存在就獲取他的第一個構造器(無參構造器),不存在就返回 Object 的構造器
        String jfrClassName = event.getClass().getPackage().getName() + ".Jfr" + event.getClass().getSimpleName();

        Class<?> eventClass = LettuceClassUtils.findClass(jfrClassName);

        if (eventClass == null) {
            constructor = Object.class.getConstructor();
        } else {
            constructor = eventClass.getDeclaredConstructors()[0];
            constructor.setAccessible(true);
        }

        synchronized (constructorMap) {
            constructorMap.put(event.getClass(), constructor);
        }
    }

    return constructor;
}

發現這塊程式碼並不是很好,每次讀都要獲取鎖,所以我做了點修改並提了一個 Pull Request: reformat getEventConstructor for JfrEventRecorder not to synchronize for each read

由此我們可以知道,一個 Event 是否有對應的 JFR Event 通過檢視是否有同路徑的以 Jfr 開頭後面跟著自己名字的類即可。目前可以發現:

  • io.lettuce.core.event.connection 包:

    • ConnectedEvent -> JfrConnectedEvent
    • ConnectEvent -> JfrConnectedEvent
    • ConnectionActivatedEvent -> JfrConnectionActivatedEvent
    • ConnectionCreatedEvent -> JfrConnectionCreatedEvent
    • ConnectionDeactivatedEvent -> JfrConnectionDeactivatedEvent
    • DisconnectedEvent -> JfrDisconnectedEvent
    • ReconnectAttemptEvent -> JfrReconnectAttemptEvent
    • ReconnectFailedEvent -> JfrReconnectFailedEvent
  • io.lettuce.core.cluster.event 包:

    • AskRedirectionEvent -> JfrAskRedirectionEvent
    • ClusterTopologyChangedEvent -> JfrClusterTopologyChangedEvent
    • MovedRedirectionEvent -> JfrMovedRedirectionEvent
    • AskRedirectionEvent -> JfrTopologyRefreshEvent
  • io.lettuce.core.event.command 包:

    CommandStartedEvent
    CommandSucceededEvent
    CommandFailedEvent
    
  • io.lettuce.core.event.metrics 包:、

    • CommandLatencyEvent -> 無

我們可以看到,當前針對指令,並沒有 JFR 監控,但是對於我們來說,指令監控反而是最重要的。我們考慮 針對指令相關事件新增 JFR 對應事件

如果對 io.lettuce.core.event.command 包下的指令事件生成對應的 JFR,那麼這個 事件數量有點太多了 (我們一個應用例項可能每秒執行好幾十萬個 Redis 指令)。所以我們傾向於針對 CommandLatencyEvent 新增 JFR 事件。

CommandLatencyEvent 包含一個 Map:

private Map<CommandLatencyId, CommandMetrics> latencies;

其中 CommandLatencyId 包含 Redis 連線資訊,以及執行的命令 。CommandMetrics 即時間統計,包含:

  • 收到 Redis 伺服器響應的時間指標 ,通過這個判斷是否是 Redis 伺服器響應慢。
  • 處理完 Redis 伺服器響應的時間指標 ,可能由於應用例項過忙導致響應一直沒有處理完,通過這個與 收到 Redis 伺服器響應的時間指標 對比判斷應用處理花的時間。

這兩個指標都包含如下資訊:

  • 最短時間
  • 最長時間
  • 百分位時間,預設是前 50%,前 90%,前 95%,前 99%,前 99.9%,對應原始碼: MicrometerOptions : public static final double[] DEFAULT_TARGET_PERCENTILES = new double[] { 0.50, 0.90, 0.95, 0.99, 0.999 };

我們想要實現針對每個不同 Redis 伺服器每個命令都能通過 JFR 檢視一段時間內響應時間指標的統計,可以這樣實現:

package io.lettuce.core.event.metrics;

import jdk.jfr.Category;
import jdk.jfr.Event;
import jdk.jfr.Label;
import jdk.jfr.StackTrace;

@Category({ "Lettuce", "Command Events" })
@Label("Command Latency Trigger")
@StackTrace(false)
public class JfrCommandLatencyEvent extends Event {
    private final int size;

    public JfrCommandLatencyEvent(CommandLatencyEvent commandLatencyEvent) {
        this.size = commandLatencyEvent.getLatencies().size();
        commandLatencyEvent.getLatencies().forEach((commandLatencyId, commandMetrics) -> {
            JfrCommandLatency jfrCommandLatency = new JfrCommandLatency(commandLatencyId, commandMetrics);
            jfrCommandLatency.commit();
        });
    }
}
package io.lettuce.core.event.metrics;

import io.lettuce.core.metrics.CommandLatencyId;
import io.lettuce.core.metrics.CommandMetrics;
import jdk.jfr.Category;
import jdk.jfr.Event;
import jdk.jfr.Label;
import jdk.jfr.StackTrace;

import java.util.concurrent.TimeUnit;

@Category({ "Lettuce", "Command Events" })
@Label("Command Latency")
@StackTrace(false)
public class JfrCommandLatency extends Event {
    private final String remoteAddress;
    private final String commandType;
    private final long count;
    private final TimeUnit timeUnit;
    private final long firstResponseMin;
    private final long firstResponseMax;
    private final String firstResponsePercentiles;
    private final long completionResponseMin;
    private final long completionResponseMax;
    private final String completionResponsePercentiles;

    public JfrCommandLatency(CommandLatencyId commandLatencyId, CommandMetrics commandMetrics) {
        this.remoteAddress = commandLatencyId.remoteAddress().toString();
        this.commandType = commandLatencyId.commandType().toString();
        this.count = commandMetrics.getCount();
        this.timeUnit = commandMetrics.getTimeUnit();
        this.firstResponseMin = commandMetrics.getFirstResponse().getMin();
        this.firstResponseMax = commandMetrics.getFirstResponse().getMax();
        this.firstResponsePercentiles = commandMetrics.getFirstResponse().getPercentiles().toString();
        this.completionResponseMin = commandMetrics.getCompletion().getMin();
        this.completionResponseMax = commandMetrics.getCompletion().getMax();
        this.completionResponsePercentiles = commandMetrics.getCompletion().getPercentiles().toString();
    }
}

這樣,我們就可以這樣分析這些事件:

首先在事件瀏覽器中,選擇 Lettuce -> Command Events -> Command Latency,右鍵使用事件建立新頁:

在建立的事件頁中,按照 commandType 分組,並且將感興趣的指標顯示到圖表中:

針對這些修改, 我也向社群提了一個 Pull Request fix #1820 add JFR Event for Command Latency

在 Spring Boot 中(即增加了 spring-boot-starter-redis 依賴),我們需要手動開啟 CommandLatencyEvent 的採集:

@Configuration(proxyBeanMethods = false)
@Import({LettuceConfiguration.class})
//需要強制在 RedisAutoConfiguration 進行自動裝載
@AutoConfigureBefore(RedisAutoConfiguration.class)
public class LettuceAutoConfiguration {
}
import io.lettuce.core.event.DefaultEventPublisherOptions;
import io.lettuce.core.metrics.DefaultCommandLatencyCollector;
import io.lettuce.core.metrics.DefaultCommandLatencyCollectorOptions;
import io.lettuce.core.resource.DefaultClientResources;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration(proxyBeanMethods = false)
public class LettuceConfiguration {
    /**
     * 每 10s 採集一次命令統計
     * @return
     */
    @Bean
    public DefaultClientResources getDefaultClientResources() {
        DefaultClientResources build = DefaultClientResources.builder()
                .commandLatencyRecorder(
                        new DefaultCommandLatencyCollector(
                                //開啟 CommandLatency 事件採集,並且配置每次採集後都清空資料
                                DefaultCommandLatencyCollectorOptions.builder().enable().resetLatenciesAfterEvent(true).build()
                        )
                )
                .commandLatencyPublisherOptions(
                        //每 10s 採集一次命令統計
                        DefaultEventPublisherOptions.builder().eventEmitInterval(Duration.ofSeconds(10)).build()
                ).build();
        return build;
    }
}

微信搜尋“我的程式設計喵”關注公眾號,每日一刷,輕鬆提升技術,斬獲各種offer