Spring Websocket в кластере tomcat

Ваше требование можно разделить на 2 подзадачи:

1. Сохранение информации о сеансе на нескольких узлах: вы можете попробовать кластеризацию Spring Sessions при поддержке Redis (см.: HttpSession с Redis). Это очень просто и уже поддерживает Spring Websockets (см.: Spring Session и WebSockets).

2. Обработка трафика веб-сокетов через несколько экземпляров tomcat: есть несколько способов сделать это.

   • The first way: Using a full-featured broker (eg: ActiveMQ) and try new feature Support multiple WebSocket servers (from: 4.2.0 RC1)
   • Второй способ: использование полнофункционального брокера и реализация распределенного UserSessionRegistry (например: использование Redis :D ). Реализация по умолчанию DefaultUserSessionRegistry использует хранилище в памяти.

Обновлено: я написал простую реализацию с использованием Redis, попробуйте, если вам интересно

Для настройки полнофункционального брокера (реле брокера) можно попробовать:

public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {

    ...

    @Autowired
    private RedisConnectionFactory redisConnectionFactory;

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableStompBrokerRelay("/topic", "/queue")
            .setRelayHost("localhost") // broker host
            .setRelayPort(61613) // broker port
            ;
        config.setApplicationDestinationPrefixes("/app");
    }

    @Bean
    public UserSessionRegistry userSessionRegistry() {
        return new RedisUserSessionRegistry(redisConnectionFactory);
    }

    ...
}

а также

import java.util.Set;

import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.messaging.simp.user.UserSessionRegistry;
import org.springframework.util.Assert;

/**
 * An implementation of {@link UserSessionRegistry} backed by Redis.
 * @author thanh
 */
public class RedisUserSessionRegistry implements UserSessionRegistry {

    /**
     * The prefix for each key of the Redis Set representing a user's sessions. The suffix is the unique user id.
     */
    static final String BOUNDED_HASH_KEY_PREFIX = "spring:websockets:users:";

    private final RedisOperations<String, String> sessionRedisOperations;

    @SuppressWarnings("unchecked")
    public RedisUserSessionRegistry(RedisConnectionFactory redisConnectionFactory) {
        this(createDefaultTemplate(redisConnectionFactory));
    }

    public RedisUserSessionRegistry(RedisOperations<String, String> sessionRedisOperations) {
        Assert.notNull(sessionRedisOperations, "sessionRedisOperations cannot be null");
        this.sessionRedisOperations = sessionRedisOperations;
    }

    @Override
    public Set<String> getSessionIds(String user) {
        Set<String> entries = getSessionBoundHashOperations(user).members();
        return (entries != null) ? entries : Collections.<String>emptySet();
    }

    @Override
    public void registerSessionId(String user, String sessionId) {
        getSessionBoundHashOperations(user).add(sessionId);
    }

    @Override
    public void unregisterSessionId(String user, String sessionId) {
        getSessionBoundHashOperations(user).remove(sessionId);
    }

    /**
     * Gets the {@link BoundHashOperations} to operate on a username
     */
    private BoundSetOperations<String, String> getSessionBoundHashOperations(String username) {
        String key = getKey(username);
        return this.sessionRedisOperations.boundSetOps(key);
    }

    /**
     * Gets the Hash key for this user by prefixing it appropriately.
     */
    static String getKey(String username) {
        return BOUNDED_HASH_KEY_PREFIX + username;
    }

    @SuppressWarnings("rawtypes")
    private static RedisTemplate createDefaultTemplate(RedisConnectionFactory connectionFactory) {
        Assert.notNull(connectionFactory, "connectionFactory cannot be null");
        StringRedisTemplate template = new StringRedisTemplate(connectionFactory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        template.afterPropertiesSet();
        return template;
    }

}

Горизонтальное масштабирование WebSockets на самом деле сильно отличается от горизонтального масштабирования приложений без сохранения состояния/отслеживания состояния, основанных только на HTTP.

Горизонтальное масштабирование HTTP-приложения без сохранения состояния: просто разверните несколько экземпляров приложения на разных компьютерах и поставьте перед ними балансировщик нагрузки. Существует довольно много различных решений для балансировки нагрузки, таких как HAProxy, Nginx и т. д. Если вы работаете в облачной среде, такой как AWS, у вас также могут быть управляемые решения, такие как Elastic Load Balancer.

Горизонтально масштабируемое HTTP-приложение с отслеживанием состояния: было бы здорово, если бы все приложения всегда оставались без состояния, но, к сожалению, это не всегда возможно. Таким образом, при работе с HTTP-приложениями с отслеживанием состояния вы должны заботиться о сеансе HTTP, который в основном представляет собой локальное хранилище для каждого отдельного клиента, где веб-сервер может хранить данные, которые хранятся в разных HTTP-запросах ( например, при работе с корзиной). Что ж, в этом случае при горизонтальном масштабировании вы должны знать, что, как я уже сказал, это хранилище LOCAL, поэтому ServerA не сможет обрабатывать сеанс HTTP, который находится на ServerB. Другими словами, если по какой-либо причине клиент 1, который обслуживается сервером А, внезапно начинает обслуживаться сервером Б, его HTTP-сессия будет потеряна (и его корзина покупок исчезнет!). Причинами могут быть отказ узла или даже развертывание. Чтобы решить эту проблему, вы не можете хранить сеансы HTTP только локально, то есть вы должны хранить их на другом внешнем компоненте. Есть несколько компонентов, которые могли бы справиться с этим, например, любая реляционная база данных, но на самом деле это было бы накладным расходом. Некоторые базы данных NoSQL могут очень хорошо справляться с таким поведением ключей и значений, например Redis. Теперь, когда сеанс HTTP хранится в Redis, если клиент начинает обслуживаться другим сервером, он извлечет сеанс HTTP клиента из Redis и загрузит его в свою память, поэтому все будет продолжать работать, и пользователь не потеряет свои данные. HTTP-сессия больше. Вы можете использовать Spring Session, чтобы легко сохранить HTTP-сессию в Redis.

Горизонтально масштабируемое приложение WebSocket. Когда соединение WebSocket установлено, сервер должен поддерживать соединение с клиентом открытым, чтобы они могли обмениваться данными в обоих направлениях. Когда клиент прослушивает адресат, такой как /topic/public.messages, мы говорим, что клиент подписан на этот адресат. В Spring, когда вы используете подход simpleBroker, подписки хранятся в памяти, так что, например, если Client1 обслуживается ServerA и хочет отправить сообщение с помощью WebSocket Client2, обслуживаемому ServerB ? Вы уже знаете ответ! Сообщение не будет доставлено Клиенту2, потому что Сервер1 даже не знает о подписке Клиента2. Итак, чтобы решить эту проблему, вам снова нужно экстернализовать подписки WebSockets. Поскольку вы используете STOMP в качестве подпротокола, вам нужен внешний компонент, который может действовать как внешний брокер STOMP. Для этого существует довольно много инструментов, но я бы предложил RabbitMQ. Теперь вы должны изменить конфигурацию Spring, чтобы подписки не сохранялись в памяти. Вместо этого он делегирует подписки внешнему брокеру STOMP. Вы можете легко добиться этого с помощью некоторых базовых конфигураций, таких как enableStompBrokerRelay. Важно отметить, что сеанс HTTP отличается от сеанса WebSocket. Использование Spring Session для хранения HTTP-сессии в Redis не имеет абсолютно никакого отношения к горизонтальному масштабированию WebSockets.

Я написал полное приложение веб-чата с Spring Boot (и многое другое), которое использует RabbitMQ в качестве полного внешнего брокера STOMP, и это общедоступно на GitHub, поэтому, пожалуйста, клонируйте его, запустите приложение на своем компьютере и просмотрите подробности кода.

Когда дело доходит до потери соединения WebSocket, Spring мало что может сделать. На самом деле переподключение должно быть запрошено клиентской стороной, реализующей, например, функцию обратного вызова переподключения (это поток рукопожатия WebSocket, клиент должен инициировать рукопожатие, а не сервер). Есть некоторые библиотеки на стороне клиента, которые могут сделать это прозрачно для вас. Это не случай SockJS. В приложении чата я также реализовал эту функцию повторного подключения.

Сохраняйте информацию о сеансе на нескольких узлах:

Предположим, у нас есть 2 хоста сервера, зарезервированные с помощью балансировщика нагрузки.

Веб-сокеты — это сокет-соединение из браузера к определенному серверу host.eg host1.

Теперь, если хост1 выйдет из строя, сокетное соединение балансировщика нагрузки — хост 1 прервется. Как spring снова откроет одно и то же соединение через веб-сокет от балансировщика нагрузки до хоста 2? браузер не должен открывать новое соединение через веб-сокет