使用Okhttp RealWebSocket出现OOM，RealCall$AsyncCall实例量过高问题求助

项目使用okhttp创建WebSocket连接，生产环境近期触发OOM。经IBM HeapAnalyzer分析堆内存，故障由okhttp3/RealCall$AsyncCall实例过多导致，实例数高达1115417个，暂无法定位根因。

运行环境 #

• jdk1.8.0_181
• okhttp3.12.1

WebSocketConnection类 #

public class WebSocketConnection extends WebSocketListener {

    private static final Logger log = LoggerFactory.getLogger(WebSocketConnection.class);

    private static int connectionCounter = 0;

    public enum ConnectionState {
        IDLE, DELAY_CONNECT, CONNECTED, CLOSED_ON_ERROR
    }

    private WebSocket webSocket = null;

    private volatile long lastReceivedTime = 0;

    private volatile ConnectionState state = ConnectionState.IDLE;
    private int delayInSecond = 0;

    private final WebsocketRequest request;
    private final Request okhttpRequest;
    private final IWebSocketWatchDog watchDog;
    private final int connectionId;
    private final boolean autoClose;
    private  String listenKey;

    private String subscriptionUrl = BinanceApiConstants.WS_API_BASE_URL;

    WebSocketConnection(String apiKey, String secretKey, SubscriptionOptions options, WebsocketRequest request,
                        WebSocketWatchDog watchDog) {
        this(apiKey, secretKey, options, request, watchDog, false);
    }

    WebSocketConnection(String apiKey, String secretKey, SubscriptionOptions options, WebsocketRequest request,
                        IWebSocketWatchDog watchDog, boolean autoClose) {
        this.connectionId = WebSocketConnection.connectionCounter++;
        this.request = request;
        this.autoClose = autoClose;

        this.okhttpRequest = StringUtils.isNotBlank(options.getUri()) ? new Request.Builder().url(options.getUri()).build()
                : new Request.Builder().url(subscriptionUrl).build();
        this.watchDog = watchDog;
        log.info("[Sub] Connection [id: " + this.connectionId + "] created for " + request.name);
    }
    WebSocketConnection(String apiKey, String secretKey,String listenKey, SubscriptionOptions options, WebsocketRequest request,
                        IWebSocketWatchDog watchDog, boolean autoClose) {
        this.connectionId = WebSocketConnection.connectionCounter++;
        this.request = request;
        this.autoClose = autoClose;
        this.listenKey = listenKey;

        this.okhttpRequest = StringUtils.isNotBlank(options.getUri()) ? new Request.Builder().url(options.getUri()).build()
                : new Request.Builder().url(subscriptionUrl).build();
        this.watchDog = watchDog;
        log.info("[Sub] Connection [id: " + this.connectionId + "] created for " + request.name);
    }
    int getConnectionId() {
        return this.connectionId;
    }

    void connect() {
        if (state == ConnectionState.CONNECTED) {
            log.info("[Sub][" + this.connectionId + "] Already connected");
            return;
        }
        webSocket = RestApiInvoker.createWebSocket(okhttpRequest, this);
    }

    void reConnect(int delayInSecond) {
        log.warn("[Sub][" + this.connectionId + "] Reconnecting after " + delayInSecond + " seconds later");
        if (webSocket != null) {
            webSocket.cancel();
            webSocket = null;
        }
        this.delayInSecond = delayInSecond;
        state = ConnectionState.DELAY_CONNECT;
    }

    void reConnect() {
        if (delayInSecond != 0) {
            delayInSecond--;
        } else {
            connect();
        }
    }

    long getLastReceivedTime() {
        return this.lastReceivedTime;
    }
    public void setLastReceivedTime(long currentTimeMillis) {
        this.lastReceivedTime=currentTimeMillis;
    }

    void send(String str) {
        boolean result = false;
        log.debug("[Send]{}", str);
        if (webSocket != null) {
            result = webSocket.send(str);
        }
        if (!result) {
            log.error("[Sub][" + this.connectionId + "] Failed to send message");
            closeOnError();
        }
    }

    @Override
    public void onMessage(WebSocket webSocket, String text) {
        super.onMessage(webSocket, text);
        lastReceivedTime = System.currentTimeMillis();

        log.debug("[On Message]:{}", text);
        try {
            JsonWrapper jsonWrapper = JsonWrapper.parseFromString(text);

            if (jsonWrapper.containKey("result") || jsonWrapper.containKey("id")) {
                // onReceiveAndClose(jsonWrapper);
            } else {
                onReceiveAndClose(jsonWrapper);
            }

        } catch (Exception e) {
            log.error("[On Message][{}]: catch exception:", connectionId, e);
            closeOnError();
        }
    }

    private void onError(String errorMessage, Throwable e) {
        if (request.errorHandler != null) {
            BinanceApiException exception = new BinanceApiException(BinanceApiException.SUBSCRIPTION_ERROR, errorMessage, e);
            request.errorHandler.onError(exception);
        }
        log.error("[Sub][" + this.connectionId + "] " + errorMessage);
    }

    private void onReceiveAndClose(JsonWrapper jsonWrapper) {
        onReceive(jsonWrapper);
        if (autoClose) {
            close();
        }
    }

    @SuppressWarnings("unchecked")
    private void onReceive(JsonWrapper jsonWrapper) {
        Object obj = null;
        try {
            obj = request.jsonParser.parseJson(jsonWrapper);
        } catch (Exception e) {
            onError("Failed to parse server's response: " + e.getMessage(), e);
        }
        request.updateCallback.onReceive(obj);
    }

    public ConnectionState getState() {
        return state;
    }

    public void close() {
        log.error("[Sub][" + this.connectionId + "] Closing normally");
        webSocket.cancel();
        webSocket = null;
        watchDog.onClosedNormally(this);
    }

    @Override
    public void onClosed(WebSocket webSocket, int code, String reason) {
        super.onClosed(webSocket, code, reason);
        if (state == ConnectionState.CONNECTED) {
            state = ConnectionState.IDLE;
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public void onOpen(WebSocket webSocket, Response response) {
        super.onOpen(webSocket, response);
        this.webSocket = webSocket;
        log.info("[Sub][" + this.connectionId + "] Connected to server");
        if (watchDog instanceof SpotWebSocketWatchDog) {
            System.out.println();
        }
        watchDog.onConnectionCreated(this);
        if (request.connectionHandler != null) {
            request.connectionHandler.handle(this);
        }
        state = ConnectionState.CONNECTED;
        lastReceivedTime = System.currentTimeMillis();
    }

    @Override
    public void onFailure(WebSocket webSocket, Throwable t, Response response) {
        onError("Unexpected error: " + t.getMessage(), t);
        closeOnError();
    }

    private void closeOnError() {
        if (webSocket != null) {
            this.webSocket.cancel();
            state = ConnectionState.CLOSED_ON_ERROR;
            log.error("[Sub][" + this.connectionId + "] Connection is closing due to error");
        }
    }

    public String getListenKey() {
        return listenKey;
    }
}

SpotWebSocketWatchDog类 #

class SpotWebSocketWatchDog implements IWebSocketWatchDog{

    private final CopyOnWriteArrayList<WebSocketConnection> TIME_HELPER = new CopyOnWriteArrayList<>();
    private final SubscriptionOptions options;
    SpotWebSocketWatchDog(SubscriptionOptions subscriptionOptions) {
        this.options = Objects.requireNonNull(subscriptionOptions);
        long t = 1_000;
        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1);
        exec.scheduleAtFixedRate(() -> {
            TIME_HELPER.forEach(connection -> {
                if (connection.getState() == WebSocketConnection.ConnectionState.CONNECTED) {
                    // Check response
                    if (options.isAutoReconnect()) {
                        long ts = System.currentTimeMillis() - connection.getLastReceivedTime();
                        if (ts > options.getReceiveLimitMs()) {
                            log.warn("[Sub SpotWebSocketWatchDog][" + connection.getConnectionId() + "] No response from server, \n send PONG to server, and update LastReceivedTime to now, jsut test!!!");
                            connection.send(Channels.userDataChannel(connection.getListenKey()));
                            connection.setLastReceivedTime(System.currentTimeMillis());
                        }
                    }
                } else if (connection.getState() == WebSocketConnection.ConnectionState.DELAY_CONNECT) {
                    connection.reConnect();
                } else if (connection.getState() == WebSocketConnection.ConnectionState.CLOSED_ON_ERROR) {
                    if (options.isAutoReconnect()) {
                        connection.reConnect(options.getConnectionDelayOnFailure());
                    }
                }
            });
        }, t, t, TimeUnit.MILLISECONDS);
        Runtime.getRuntime().addShutdownHook(new Thread(exec::shutdown));
    }

    @Override
    public void onConnectionCreated(WebSocketConnection connection) {
        TIME_HELPER.addIfAbsent(connection);
    }
    @Override
    public void onClosedNormally(WebSocketConnection connection) {
        TIME_HELPER.remove(connection);
    }

    private static volatile SpotWebSocketWatchDog instance;
    public static SpotWebSocketWatchDog getInstance(SubscriptionOptions options) {
        if (null == instance) {
            synchronized (SpotWebSocketWatchDog.class) {
                if (null == instance) {
                    instance = new SpotWebSocketWatchDog(options);
                }
            }
        }
        return SpotWebSocketWatchDog.instance;
    }

}

WebSocketStreamClientImpl类 #

public class WebSocketStreamClientImpl implements SubscriptionClient {

    private final SubscriptionOptions options;

    private IWebSocketWatchDog watchDog;
    private IWebSocketWatchDog watchSpotDog;
    private IWebSocketWatchDog watchSpotQuotesDog;

    private final WebsocketRequestImpl requestImpl;

    private final List<WebSocketConnection> connections = new LinkedList<>();

    private final String apiKey;

    private final String secretKey;

    WebSocketStreamClientImpl(String apiKey, String secretKey, SubscriptionOptions options) {
        this.apiKey = apiKey;
        this.secretKey = secretKey;
        this.watchDog = null;
        this.watchSpotDog = null;
        this.watchSpotQuotesDog = null;
        this.options = Objects.requireNonNull(options);

        this.requestImpl = new WebsocketRequestImpl();
    }

    private <T> void createConnectionSpot(WebsocketRequest<T> request, boolean autoClose, String listenKey) {
        if (watchSpotDog == null) {
            watchSpotDog = SpotWebSocketWatchDog.getInstance(options);
        }
        WebSocketConnection connection = new WebSocketConnection(apiKey, secretKey, listenKey, options, request, watchSpotDog,
                autoClose);
        if (autoClose == false) {
            connections.add(connection);
        }
        connection.connect();
    }

    private <T> void createConnectionSpot(WebsocketRequest<T> request, boolean autoClose) {
        if (watchSpotQuotesDog == null) {
            watchSpotQuotesDog =  SpotQuotesWebSocketWatchDog.getInstance(options);
        }
        WebSocketConnection connection = new WebSocketConnection(apiKey, secretKey, options, request, watchSpotQuotesDog,
                autoClose);
        if (autoClose == false) {
            connections.add(connection);
        }
        connection.connect();
    }

    private <T> void createConnection(WebsocketRequest<T> request, boolean autoClose) {
        if (watchDog == null) {
            watchDog = new WebSocketWatchDog(options);
        }
        WebSocketConnection connection = new WebSocketConnection(apiKey, secretKey, options, request, watchDog,
                autoClose);
        if (autoClose == false) {
            connections.add(connection);
        }
        connection.connect();
    }

    private <T> void createConnection(WebsocketRequest<T> request) {
        createConnection(request, false);
    }

    private <T> void createConnectionSpot(WebsocketRequest<T> request, String listenKey) {
        createConnectionSpot(request, false, listenKey);
    }

    private <T> void createConnectionSpot(WebsocketRequest<T> request) {
        createConnectionSpot(request, false);
    }

    @Override
    public void unsubscribeAll() {
        for (WebSocketConnection connection : connections) {
            watchDog.onClosedNormally(connection);
            watchSpotDog.onClosedNormally(connection);
            watchSpotQuotesDog.onClosedNormally(connection);
            connection.close();
        }
        connections.clear();
    }
 @Override
    public void subCashBalanceUpdateEvent(String cashListenKey,
                                          SubscriptionListener<CashBalanceUpdate> subscriptionListener,
                                          SubscriptionErrorHandler errorHandler) {
        createConnection(
                requestImpl.subCashBalanceUpdateEvent(cashListenKey, subscriptionListener, errorHandler));
    }

   
    @Override
    public void subscribeSpotDataEvent(String listenKey, SubscriptionListener<SpotDataUpdateEvent> callback, SubscriptionErrorHandler errorHandler) {
        createConnectionSpot(
                requestImpl.subscribeSpotDataEvent(listenKey, callback, errorHandler), listenKey);
    }

    @Override
    public void subscribeSpotBookDepthEvent(String symbol, Integer limit, SubscriptionListener<SpotOrderBookEvent> subscriptionListener, SubscriptionErrorHandler errorHandler) {
        createConnectionSpot(
                requestImpl.subscribeSpotBookDepthEvent(symbol, limit, subscriptionListener, errorHandler));
    }

}

• 无限重连导致请求爆炸：WatchDog每秒调度一次，当网络异常/服务端不可用导致连接连续失败时，每秒会创建一个新的WebSocket连接请求，每个请求对应一个RealCall$AsyncCall实例，没有重试上限的情况下实例数会无限累加。
• OkHttpClient未复用：若RestApiInvoker.createWebSocket内部每次创建新的OkHttpClient实例，每个实例持有的请求队列、线程池资源不会被主动回收，会加剧内存泄漏。
• 状态流转无原子性保障：ConnectionState仅用volatile修饰，多线程（WatchDog调度线程、okhttp回调线程）并发修改状态时会出现错乱，可能重复触发connect()调用，重复创建请求实例。
• 失效连接未清理：仅正常关闭时才会将连接从WatchDog的连接列表中移除，异常状态的连接会一直留在列表中被反复调度重连，持续创建新的请求实例。

• 新增重连次数上限：给每个连接增加重试次数统计，超过阈值后将连接从WatchDog列表中移除，停止重试，避免无限创建请求。
• 复用OkHttpClient单例：全局仅初始化一个OkHttpClient实例，所有WebSocket请求复用该实例，避免重复创建资源。
• 保障状态流转原子性：给连接状态修改逻辑加锁，避免并发修改导致的状态错乱，connect()方法执行前增加状态二次校验，避免重复调用。
• 增加异常连接清理逻辑：连接连续失败达到阈值、或长期处于非活跃状态时，主动从WatchDog列表和客户端连接列表中移除，释放关联资源。
• 新建连接前强制清理旧资源：每次调用connect()前，先强制cancel旧的WebSocket实例，清空关联的请求资源。

内容的提问来源于stack exchange，提问作者bob bob