Веб-сокеты Java в Tomcat вызывают очень высокое использование памяти и ООО ⇐ JAVA

[Anonymous]

Я пишу приложение, которое формирует около 4 тысяч соединений и отправляет данные. но почему-то он потребляет память в диапазоне более 700 ГБ. каков идеальный способ оптимизации? любое предложение по сборщику мусора, которое может помочь? даже принудительное использование GC при каждой отправке сообщения не поможет.

Цель — протестировать постоянное длинное соединение с использованием веб-сокетов в больших масштабах.
импортировать java.io.IOException; импортировать java.util.HashMap; импортировать java.util.Timer; импортировать java.util.TimerTask; импортировать java.util.concurrent.CompletableFuture; импортировать java.util.concurrent.TimeUnit; импортировать javax.websocket.Session; импортировать org.springframework.scheduling.annotation.Async; импортировать org.springframework.stereotype.Component; импортировать org.springframework.web.socket.CloseStatus; импортировать org.springframework.web.socket.TextMessage; импортировать org.springframework.web.socket.WebSocketSession; импортировать org.springframework.web.socket.adapter.NativeWebSocketSession; импортировать org.springframework.web.socket.handler.TextWebSocketHandler; импортировать n1wsperf.server.ServerTelemetry.ServerMetricType; @Компонент публичный класс TextSocketHandler расширяет TextWebSocketHandler { частный изменчивый статический HashMap activeSessions = new HashMap() ; частный статический финал int maxSizeInKB = 61440; // 60 МБ или 480 МБ частный статический окончательный TextMessage _data = DataService.getData(); частная статическая строка modeStr = «Запрос-ответ»; общественная статическая сила Initialize() { Таймер таймер = новый таймер («ActiveSessionTimer»); Задача TimerTask = новый TimerTask() { @Override общественный недействительный запуск () { ServerTelemetry.TrackMetric(ServerMetricType.ACTIVESESSION, activeSessions.size()); System.out.println("Активные соединения: " + activeSessions.size()); } }; CompletableFuture.runAsync( () -> { timer.schedule(task, 1000, 60000L); }); } общественный TextSocketHandler() { modeStr = ServerConfig.getServerRunConfig().DuplexMode? «Дуплекс»: «Запрос-Ответ»; } @Асинхронный @Override public void afterConnectionEstablished (сеанс WebSocketSession) выдает исключение { пытаться { session.setTextMessageSizeLimit(maxSizeInKB * 1024); // 60 МБ или 480 МБ Строка sessionId = session.getId(); если (экземпляр сеанса NativeWebSocketSession) { окончательный сеанс ownSession = ((NativeWebSocketSession)session).getNativeSession(Session.class); если (nativeSession!= null) { NativeSession.getUserProperties() .put("org.apache.tomcat.websocket.BLOCKING_SEND_TIMEOUT", 60_000L); } } session.sendMessage(новый TextMessage(sessionId)); activeSessions.putIfAbsent(session.getId(), session); ServerTelemetry.TrackMetric(ServerMetricType.NEWSESSION); TraceLogger.Info("SocketHandler", String.format("Сеанс(новый): %s: Режим: %s: Новое соединение установлено с клиентом %s", session.getId(), modeStr,session.getRemoteAddress()), истинный); Запустить run = ServerConfig.getServerRunConfig(); если (run.DuplexMode) { startDuplexCommunicationAsync (sessionId, run); // неблокирующий } TraceLogger.Info("SocketHandler", String.format("Сеанс (подробно: %s): %s: Режим: %s Идет связь с клиентом %s", run.Verbose, session.getId(), modeStr, session.getRemoteAddress()), true); } catch (Исключение е) { ServerTelemetry.TrackMetric(ServerMetricType.FAILEDSESSION); ServerTelemetry.TrackException(e); TraceLogger.Error("SocketHandler", String.format("Сеанс: %s Режим: %s Исключение: %s", session.getId(), modeStr,e.getMessage())); } окончательно { Система.gc(); } } @Асинхронный @Override public void afterConnectionClosed (сеанс WebSocketSession, статус CloseStatus) выдает исключение { пытаться{ вар sessionId = session.getId(); вар clientAdr = session.getRemoteAddress(); ServerTelemetry.TrackMetric(ServerMetricType.CLOSEDSESSION); TraceLogger.Info("SocketHandler", String.format("Сеанс: %s Режим: %s: закрытое соединение с клиентом %s", sessionId, modeStr,clientAdr), true); если (activeSessions.keySet().contains(sessionId)){ activeSessions.remove(sessionId); ServerTelemetry.TrackMetric(ServerMetricType.ACTIVESESSION, activeSessions.size()); }} окончательно { Система.gc(); } } @Асинхронный @Override protected void handleTextMessage (сеанс WebSocketSession, сообщение TextMessage) выдает исключение { пытаться { Строка sessionId = session.getId(); WebSocketSession wsSession = activeSessions.get(sessionId); длинный старт = System.nanoTime(); //ServerTelemetry.TrackMetric(ServerMetricType.REQUEST); Строковая полезная нагрузка = message.getPayload(); //ServerTelemetry.TrackMetric(ServerMetricType.REQUESTSIZE, payload.length()); // Байты TraceLogger.Info("SocketHandler", String.format("Сеанс: %s: Режим: %s: получено %s байт от клиента %s", sessionId, modeStr,payload.length(), wsSession.getRemoteAddress())) ; if (!ServerConfig.getServerRunConfig().DuplexMode) { CompletableFuture.runAsync( () -> { пытаться { sendMessage (wsSession, старт); } catch (IOException | InterruptedException e) { TraceLogger.Error("SocketHandler", String.format("Сеанс: %s Режим: %s Исключение: %s", sessionId, modeStr,e.getMessage())); } }).получать(); } длинный конец = System.nanoTime(); double reqDurationms = TimeUnit.NANOSECONDS.toMillis(конец - начало); // ServerTelemetry.TrackMetric(ServerMetricType.ACTIVESESSION, activeSessions.size()); // ServerTelemetry.TrackMetric(ServerMetricType.DURATION, reqDurationms); } catch (Исключение е) { ServerTelemetry.TrackMetric(ServerMetricType.FAILEDREQUEST); ServerTelemetry.TrackException(e); TraceLogger.Error("SocketHandler", String.format("Сеанс: %s Режим: %s Исключение: %s", session.getId(), modeStr,e.getMessage())); } окончательно { Система.gc(); } } @Асинхронный @Override public void handleTransportError (сеанс WebSocketSession, исключение Throwable) выдает исключение { пытаться{ WebSocketSession wsSession = activeSessions.get(session.getId()); ServerTelemetry.TrackMetric(ServerMetricType.TRANSPORTERROR); TraceLogger.Error("SocketHandler", String.format("Сеанс: %s: Режим: %s Ошибка: %s", wsSession.getId(), modeStr,Exception.getMessage())); if (activeSessions.keySet().contains(wsSession.getId()))){ activeSessions.remove(wsSession.getId()); ServerTelemetry.TrackMetric(ServerMetricType.ACTIVESESSION, activeSessions.size()); }} окончательно { Система.gc(); } } Private void startDuplexCommunicationAsync (String sessionId, Run run) выдает IOException, InterruptedException { CompletableFuture.runAsync(() -> { длинный сеансStart = System.nanoTime(); WebSocketSession wsSession = null; for (long current = System.nanoTime(); current < sessionStart + TimeUnit.SECONDS.toNanos(run.KeepAlive_Sec); ){ пытаться { wsSession = activeSessions.get(sessionId); sendMessage(wsSession, System.nanoTime()); текущий = System.nanoTime(); } catch (IOException | InterruptedException e) { TraceLogger.Error("SocketHandler", String.format("Сеанс: %s Режим: %s Исключение: %s", sessionId, modeStr,e.getMessage())); } } если (wsSession != ноль && wsSession.isOpen()){ пытаться { wsSession.close(CloseStatus.NORMAL); TraceLogger.Info("SocketHandler", String.format("Сеанс: %s Режим: %s: Закрытие сеанса по истечении настроенной продолжительности активности (%s секунд), отправленный сеанс близко к клиенту %s", sessionId, modeStr, run.KeepAlive_Sec, wsSession.getRemoteAddress()), true); } catch (IOException e) { TraceLogger.Error("SocketHandler", String.format("Сеанс: %s Режим: %s Исключение: %s", sessionId, modeStr,e.getMessage())); } } }); } Private void sendMessage (сеанс WebSocketSession, длительный запуск) выдает IOException, InterruptedException { если (!session.isOpen()) { throw new IOException("Сессия не находится в открытом состоянии"); } вар runCfg = ServerConfig.getServerRunConfig(); пытаться { session.sendMessage(_data); //ServerTelemetry.TrackMetric(ServerMetricType.RESPONSESIZE, _data.getPayloadLength()); // Байты TraceLogger.Info("SocketHandler", String.format("Сеанс: %s: Режим: %s: %s байт отправлено клиенту %s", session.getId(), modeStr, _data.getPayloadLength(), session.getRemoteAddress ())); длинный конец = System.nanoTime(); long restMillis = runCfg.RequestInterval_ms - TimeUnit.NANOSECONDS.toMillis(конец - начало); если (остатьсяМиллис > 0) { Thread.sleep(remainMillis); } } catch (IOException e) { TraceLogger.Error("SocketHandler", String.format("Сеанс: %s: Режим: %s: Исключение: %s", session.getId(), modeStr,e.getMessage())); } окончательно { Система.gc(); } } } [*]Различные виды сборщика мусора, параллельные с 50 % времени сборки мусора, все еще выполняются, доступная память обнулилась. [*]Удаление ведения журнала