【发布时间】：2021-09-29 08:43:53
【问题描述】：
我在项目中使用okhttp创建websocket连接 今天生产环境发OOM。使用IBM HeapAnalyzer分析堆信息，发现是okhttp3/RealCall$AsyncCall引起的。多达 1,115,417 我不知道为什么我希望得到帮助
jdk1.8.0_181 好的http3.12.1
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;
}
// log.info("[Sub][" + this.connectionId + "] Connecting...");
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);
}
// try {
request.updateCallback.onReceive(obj);
// } catch (Exception e) {
// onError("Process error: " + e.getMessage() + " You should capture the exception in your error handler", e);
// }
}
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;
}
}
我的现场看门狗
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;
}
}
我的客户
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));
}
}
【问题讨论】：
标签： okhttp