Зависшие подписки

Зависшие подписки в STOMP.js возникают в ситуациях, когда клиент больше не использует подписку, но сервер продолжает считать её активной. Это приводит к накоплению неосвобождённых ресурсов, дублированию сообщений, росту сетевой нагрузки и постепенному ухудшению производительности приложения.

Подобные проблемы особенно часто появляются в:

  • SPA-приложениях;
  • React/Vue/Angular-компонентах;
  • динамических вкладках;
  • системах уведомлений;
  • чатах;
  • дашбордах с обновлением данных в реальном времени.

Наиболее распространённые причины:

Повторная подписка без unsubscribe

client.subscribe('/topic/orders', message => {
    console.log(message.body);
});

При повторном выполнении этого кода создаётся новая подписка, а старая остаётся активной.

Если компонент перерисовывается многократно, количество подписок начинает быстро расти.

Потеря ссылки на подписку

Метод subscribe() возвращает объект подписки:

const subscription = client.subscribe('/topic/orders', callback);

Если ссылка потеряна, вызвать unsubscribe() становится невозможно.

Ошибочный вариант:

function init() {
    client.subscribe('/topic/orders', callback);
}

Правильный вариант:

let subscription;

function init() {
    subscription = client.subscribe('/topic/orders', callback);
}

function destroy() {
    subscription.unsubscribe();
}

Повторное подключение после reconnect

После автоматического переподключения многие разработчики повторно вызывают регистрацию подписок:

client.onConn ect = () => {
    client.subscribe('/topic/orders', callback);
};

Если предыдущие подписки не были очищены корректно, сервер может сохранить старые соединения и подписки.

В результате одно сообщение начинает приходить несколько раз.

Некорректное завершение WebSocket

При аварийном закрытии вкладки браузера или потере сети STOMP-сессия может завершиться не полностью.

Сервер продолжает считать подписку активной до истечения таймаута heartbeat.

Подписки внутри обработчиков

Критическая ошибка:

client.subscribe('/topic/main', message => {

    client.subscribe('/topic/inner', inner => {
        console.log(inner.body);
    });

});

Каждое сообщение создаёт новую вложенную подписку.

Через несколько минут приложение может иметь сотни активных подписчиков.

Как устроены подписки в STOMP.js

Каждая подписка получает уникальный идентификатор:

const sub = client.subscribe(destination, callback);

Внутри STOMP формируется кадр:

SUBSCRIBE
id:sub-0
destination:/topic/orders

При отмене подписки отправляется:

UNSUBSCRIBE
id:sub-0

Если UNSUBSCRIBE не отправлен, сервер продолжает хранить подписку.

Симптомы зависших подписок

Дублирование сообщений

Одно сообщение приходит несколько раз:

Order upd ated
Order upd ated
Order upd ated

Чем дольше работает приложение, тем больше дублей появляется.

Рост использования памяти

Каждая подписка хранит:

  • callback;
  • ссылки на замыкания;
  • внутренние структуры STOMP;
  • ссылки на DOM;
  • данные контекста компонентов.

При утечке подписок память браузера постепенно увеличивается.

Рост сетевого трафика

Сервер продолжает отправлять сообщения всем неактивным подписчикам.

Это особенно критично при:

  • биржевых данных;
  • телеметрии;
  • игровых событиях;
  • потоковых обновлениях.

Повышенная нагрузка на сервер

Broker хранит:

  • таблицы подписок;
  • маршруты сообщений;
  • heartbeat-состояния;
  • очереди доставки.

Тысячи «мертвых» подписок могут перегрузить сервер.

Управление жизненным циклом подписок

Базовый шаблон

let subscription = null;

function subscribe() {

    if (subscription) {
        subscription.unsubscribe();
    }

    subscription = client.subscribe(
        '/topic/orders',
        message => {
            console.log(message.body);
        }
    );
}

function unsubscribe() {

    if (subscription) {
        subscription.unsubscribe();
        subscription = null;
    }
}

Массовое управление подписками

В крупных приложениях подписок может быть десятки или сотни.

Хранение подписок в Map

const subscriptions = new Map();

function addSubscription(name, destination, callback) {

    removeSubscription(name);

    const sub = client.subscribe(destination, callback);

    subscriptions.se t(name, sub);
}

function removeSubscription(name) {

    const sub = subscriptions.get(name);

    if (sub) {
        sub.unsubscribe();
        subscriptions.delete(name);
    }
}

Полная очистка

function clearSubscriptions() {

    subscriptions.forEach(sub => {
        sub.unsubscribe();
    });

    subscriptions.clear();
}

Защита от дублирования подписок

Проверка существования

if (!subscriptions.has('orders')) {

    const sub = client.subscribe(
        '/topic/orders',
        callback
    );

    subscriptions.se t('orders', sub);
}

Использование флагов состояния

let subscribed = false;

function init() {

    if (subscribed) {
        return;
    }

    subscribed = true;

    client.subscribe('/topic/orders', callback);
}

Зависшие подписки при reconnect

Ошибочный сценарий

client.onConn ect = () => {

    client.subscribe('/topic/orders', callback);

};

При каждом reconnect создаётся новая подписка.

Безопасный вариант

let orderSubscription = null;

client.onConn ect = () => {

    if (orderSubscription) {
        orderSubscription.unsubscribe();
    }

    orderSubscription = client.subscribe(
        '/topic/orders',
        callback
    );
};

Автоматическое восстановление подписок

В сложных приложениях удобно хранить конфигурацию подписок отдельно.

const subscriptionDefinitions = [
    {
        name: 'orders',
        destination: '/topic/orders',
        callback: onOrders
    },
    {
        name: 'users',
        destination: '/topic/users',
        callback: onUsers
    }
];

Восстановление:

const activeSubscriptions = new Map();

function restoreSubscriptions() {

    subscriptionDefinitions.forEach(def => {

        const sub = client.subscribe(
            def.destination,
            def.callback
        );

        activeSubscriptions.se t(def.name, sub);
    });
}

Очистка подписок перед disconnect

Правильная последовательность

function shutdown() {

    clearSubscriptions();

    client.deactivate();
}

Опасный вариант

client.deactivate();

clearSubscriptions();

После закрытия соединения UNSUBSCRIBE может не отправиться.

Работа с подписками в React

Типичная утечка

useEffect(() => {

    client.subscribe('/topic/orders', callback);

}, []);

Подписка создаётся, но не удаляется.

Корректный вариант

useEffect(() => {

    const sub = client.subscribe(
        '/topic/orders',
        callback
    );

    return () => {
        sub.unsubscribe();
    };

}, []);

Работа с подписками в Vue

export default {

    mounted() {

        this.subscription = client.subscribe(
            '/topic/orders',
            this.onMessage
        );
    },

    beforeUnmount() {

        if (this.subscription) {
            this.subscription.unsubscribe();
        }
    }
};

Работа с подписками в Angular

private subscription: StompSubscription;

ngOnInit(): void {

    this.subscription = this.client.subscribe(
        '/topic/orders',
        this.onMessage
    );
}

ngOnDestroy(): void {

    if (this.subscription) {
        this.subscription.unsubscribe();
    }
}

Подписки и утечки памяти

Подписка может удерживать ссылки на:

  • DOM-элементы;
  • сервисы;
  • большие массивы данных;
  • state-компонентов;
  • WebGL-ресурсы;
  • Canvas-контексты.

Даже если компонент удалён, зависшая подписка мешает сборщику мусора освободить память.

Диагностика зависших подписок

Логирование подписок

function trackedSubscribe(name, destination, callback) {

    console.log('SUBSCRIBE:', name);

    const sub = client.subscribe(destination, callback);

    return {
        unsubscribe() {
            console.log('UNSUBSCRIBE:', name);
            sub.unsubscribe();
        }
    };
}

Подсчёт активных подписок

console.log(
    'Subscriptions:',
    subscriptions.size
);

Отслеживание reconnect

client.onConn ect = () => {
    console.log('CONNECTED');
};

client.onDisconn ect = () => {
    console.log('DISCONNECTED');
};

Серверные последствия зависших подписок

RabbitMQ

При использовании STOMP поверх RabbitMQ подписки могут создавать:

  • очереди;
  • consumer-сессии;
  • routing bindings.

Неочищенные подписки приводят к росту потребления памяти broker’ом.

ActiveMQ

В ActiveMQ зависшие подписчики способны:

  • накапливать pending messages;
  • удерживать durable subscriptions;
  • создавать ghost consumers.

Spring WebSocket

В Spring-приложениях зависшие подписки могут оставаться в:

  • SimpleBroker;
  • SimpUserRegistry;
  • user destination cache.

Heartbeat и очистка зависших соединений

Heartbeat помогает серверу обнаруживать неактивных клиентов.

Настройка:

client.heartbeatIncoming = 4000;
client.heartbeatOutgoing = 4000;

Если heartbeat отсутствует слишком долго, сервер закрывает соединение и удаляет подписки.

Durable subscriptions

Некоторые брокеры поддерживают durable-подписки.

Они сохраняются даже после отключения клиента.

Ошибка управления durable subscriptions может привести к:

  • накоплению сообщений;
  • бесконечному росту очередей;
  • переполнению диска.

Защита от повторной регистрации

Централизованный менеджер

class SubscriptionManager {

    constructor(client) {
        this.client = client;
        this.subscriptions = new Map();
    }

    subscribe(name, destination, callback) {

        this.unsubscribe(name);

        const sub = this.client.subscribe(
            destination,
            callback
        );

        this.subscriptions.set(name, sub);

        return sub;
    }

    unsubscribe(name) {

        const sub = this.subscriptions.get(name);

        if (sub) {
            sub.unsubscribe();
            this.subscriptions.delete(name);
        }
    }

    clear() {

        this.subscriptions.forEach(sub => {
            sub.unsubscribe();
        });

        this.subscriptions.clear();
    }
}

Проверка состояния соединения

Перед созданием подписки желательно проверять соединение:

if (client.connected) {

    client.subscribe(
        '/topic/orders',
        callback
    );
}

Использование уникальных идентификаторов

STOMP поддерживает собственные id подписок:

client.subscribe(
    '/topic/orders',
    callback,
    {
        id: 'orders-subscription'
    }
);

Это облегчает диагностику на сервере.

Очистка подписок при смене страницы

window.addEventListener('beforeunload', () => {

    clearSubscriptions();

    client.deactivate();
});

Race condition при переподключении

Опасный сценарий:

client.deactivate();

client.activate();

Если предыдущие подписки ещё не удалены, возможно дублирование.

Безопасный вариант

await client.deactivate();

client.activate();

Мониторинг количества подписок

Полезно отслеживать:

  • число подписок на клиента;
  • число reconnect;
  • частоту subscribe/unsubscribe;
  • количество duplicate messages;
  • использование памяти.

Признаки архитектурной проблемы

Если приложение постоянно создаёт и уничтожает подписки, это может означать:

  • неправильную структуру компонентов;
  • избыточную реактивность;
  • отсутствие кэширования;
  • неверную организацию состояния приложения;
  • ошибки жизненного цикла UI.

Практика безопасной работы с подписками

Основные правила

  • каждая подписка должна иметь unsubscribe;
  • подписки должны храниться централизованно;
  • reconnect не должен создавать дубликаты;
  • перед disconnect требуется очистка;
  • подписки компонентов должны уничтожаться вместе с компонентами;
  • нельзя создавать подписки внутри обработчиков сообщений;
  • необходимо контролировать reconnect-циклы;
  • heartbeat должен быть включён;
  • следует логировать жизненный цикл подписок.

Типовая архитектура

class RealtimeService {

    constructor(client) {

        this.client = client;

        this.subscriptions = new Map();
    }

    subscribe(name, destination, callback) {

        this.unsubscribe(name);

        const sub = this.client.subscribe(
            destination,
            callback
        );

        this.subscriptions.set(name, sub);
    }

    unsubscribe(name) {

        const sub = this.subscriptions.get(name);

        if (sub) {
            sub.unsubscribe();
            this.subscriptions.delete(name);
        }
    }

    disconnect() {

        this.subscriptions.forEach(sub => {
            sub.unsubscribe();
        });

        this.subscriptions.clear();

        this.client.deactivate();
    }
}
Оглавление