标签: spring-websocket

在客户端我使用Stomp进行websocket连接,服务器端我正在使用Spring 4它在客户端我做了配置

var socket = new SockJS(urlBase + "/" + contextroot+'/hello');
stompClient = Stomp.over(socket);

下面的代码每2秒执行一次,将数据发送到服务器

stompClient.send('/app/sendRequest/'+indexVal, {}, 
JSON.stringify({index : simIndex}));

服务器响应以下队列

stompClient.subscribe('/queue/response', processResponseObj);

在服务器端配置完成

<websocket:message-broker application-destination-prefix="/app">
   <!--<websocket:transport send-timeout="15000" message-size="1051648" send-buffer-size="1051648"/> -->
   <websocket:stomp-endpoint path="/hello">
       <websocket:sockjs />
   </websocket:stomp-endpoint>
   <websocket:stomp-broker-relay prefix="/topic, /queue"  />

    <websocket:message-converters>
        <beans:bean class="org.springframework.messaging.converter.MappingJackson2MessageConverter">
        <beans:property name="objectMapper" ref="objectMapper" />
       </beans:bean>
    </websocket:message-converters>

    </websocket:message-broker>

消息响应在服务器端

 messagingTemplate.convertAndSend("/queue/response",obj);

几秒钟后,我收到错误消息

2014-10-24 16:39:33,869 ERROR et.messaging.SubProtocolWebSocketHandler: 330 - Terminating session id 'dkbzrkxp'
org.springframework.web.socket.handler.SessionLimitExceededException: The send buffer size 1147188 bytes for session 'dkbzrkxp exceeded the allowed limit 1051648
    at org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator.sessionLimitReached(ConcurrentWebSocketSessionDecorator.java:162) …

我有一个应用程序，其中客户端使用 websockets 连接到运行 Spring Boot Tomcat 的服务器。我的问题是服务器是否有办法检测由于网络丢失而导致的客户端断开连接。

谢谢。

我正在构建一个小型 websocket 项目并由 JWT 令牌机制保护，我想将 websocket 会话存储在 redis 而不是本地内存中。

@Override
protected void configureStompEndpoints(StompEndpointRegistry registry) {
    registry
            .addEndpoint("/hello")
            .addInterceptors(new HttpSessionHandshakeInterceptor() {
                @Override
                public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
                        WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
                    if (request instanceof ServletServerHttpRequest) {
                        ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request;
                        HttpSession session = servletRequest.getServletRequest().getSession();
                        attributes.put("sessionId", session.getId());
                    }
                    return true;
                }
            })
            .withSockJS()
            .setSessionCookieNeeded(true);
}

我在我的 redis 配置中使用 @EnableRedisHttpSession 并且上面的源代码安全地将 http 会话存储在 redis 中，但即使我刷新了我的 redis 数据库，我仍然可以向我连接的客户端发送和接收我的消息。

我的猜测是 HttpSession 与 WebSocket 会话不同。如果您使用基于令牌的身份验证，我应该如何“智能地”要求 spring 在 redis 中存储和维护与 websocket …

我正在尝试将自定义标头添加到客户端在第一次连接时收到的 STOMP 'CREATED' 消息中。这是使用 STOMP JavaScript 连接到 WebSocket 的函数：

function connect() {
    socket = new SockJS('/chat');
    stompClient = Stomp.over(socket);
    stompClient.connect('', '', function(frame) {
      whoami = frame.headers['user-name'];
      console.log(frame);
      stompClient.subscribe('/user/queue/messages', function(message) {
          console.log("MESSAGE RECEIVED:");
          console.log(message);

        showMessage(JSON.parse(message.body));
      });
      stompClient.subscribe('/topic/active', function(activeMembers) {
        showActive(activeMembers);
      });
    });
  }

此函数将以下内容打印到浏览器的控制台：

body: ""
command: "CONNECTED"
headers: Object
    heart-beat: "0,0"
    user-name: "someuser"
    version: "1.1"

我想添加自定义标题，因此输出必须如下所示：

body: ""
command: "CONNECTED"
headers: Object
    heart-beat: "0,0"
    user-name: "someuser"
    version: "1.1"
    custom-header: "foo"

我的 Spring Boot 应用程序中有以下 WebSocket 配置。

WebSocketConfig.java

@Configuration
@EnableWebSocketMessageBroker
public …

我设法WebSocketHandler使用 Spring 5 Reactive WebSocket 支持创建一个（第 23.2.4 章）。接收和发送一切正常。但是，我不知道如何检测客户端断开连接。HttpServerWSOperations当调试客户端断开连接时，它会在类（包）的服务器端某个地方停止netty.http.server，在那里它确实检测到CloseWebSocketFrame.

对于如何处理客户端断开连接有什么建议吗？

问题
有没有办法全局处理MessageDeliveryExceptionSpring WebSocket 模块中由错误（通常权限不足）引起的Spring Messaging ？

用例
我已经通过 STOMP 实现了 Spring WebSockets 以支持我的 web 应用程序中的 ws 连接。为了保护 websocket 端点，我创建了授权用户在 STOMP CONNECT 时间启动 STOMP 会话的拦截器（如 Spring 文档中 22.4.11 部分的建议）：

@Component
public class StompMessagingInterceptor extends ChannelInterceptorAdapter {

    // Some code not important to the problem

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor headerAccessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);

        switch (headerAccessor.getCommand()) {
            // Authenticate STOMP session on CONNECT using jwt token passed as a STOMP login header - it's …

我正在努力解决性能问题超过两个月,我无法解决它.因此,我想问一下问题可能是什么.

问题

我们使用Spring Websocket开发一个应用程序,其中客户端订阅其用户队列(/user/exchange/amq.direct/update)以接收它们的更新.我设置了JMeter测试,该测试订阅用户队列并每4秒发送一条消息,会话长度为12分钟.当运行大约300个线程(大约4500个请求/分钟结束)时,服务器的响应时间急剧增加到超过6秒(我认为在我的JMeter测试中超时).

测试在单独的机器上运行,应用程序在Linux(Debian)机器上运行,没有其他正在运行的应用程序.

我已经尝试过了什么

• 使用内部消息代理交换外部消息代理(rabbitMQ).有趣的是,对于内部代理,我可以处理多达2000个线程,但后来我进入相同的超时.
• 我删除了每个业务逻辑,只发送一个简单的字符串作为STOMP消息
• 我尝试了不同的配置,为ClientInboundChannel和ClientOutboundChannel
• 我试过不同SendBufferSizeLimit的WebSocketMessageBrokerConfigurer
• 为了验证我没有网络问题,我确实测量了传入消息的时间,ClientInboundChannel直到它ClientOutboundChannel通过实现ChannelInterceptors它们发送出去.这证实答案确实需要超过6秒才能发送出去.
• 在websocket的响应时间变得非常糟糕的时候(哪个btw已经影响了新线程的STOMP连接调用),我对传统的http请求没有任何问题.
• 我的航班记录显示CPU没有问题,并且在40%的峰值时,RAM也低于阈值,大约为1.5 GB.
• 我也看不到线程争用或热门方法.
• 我没有用完文件描述符.
• 垃圾收集没问题.我在7分钟内只进行了4次运行,GC时间约为200-300 ms

由于我已经在这个话题上工作了很长一段时间,我可能忘记了很多我也尝试过的事情,所以请不要犹豫,开始谈话.

我希望你能给我更多灵感来寻找问题.

更新:

我想通了,我可以增加的cachelimit DefaultSubscriptionRegistry这是由使用SimpleMessageBroker.这导致只需使用该应用程序处理数千名用户.所以看起来这是RabbitMQ的一个问题.我用PerfTest工具测试了我们的RabbitMQ,通过测试我没有任何问题.但是测试也没有使用STOMP插件,而是使用amqp.

那么Spring Websocket中的外部消息代理有任何类似的限制吗？

更新2:

我能够通过示例应用程序重现该问题,您可以在github上找到它:https: //github.com/mld-ger/spring-websocket-performance-issue

此外,因为我认为这可能是春天的一个错误我打开了一张票:https://jira.spring.io/browse/SPR-16950

我正在尝试结合 Spring-Security 来了解有关 Spring-Websocket 的更多信息，并且正在尝试Spring 文档中的示例。

在创建我的课程WebSecurityConfig并从AbstractWebSocketMessageBrokerConfigurer我那里扩展后，我被告知AbstractWebSocketMessageBrokerConfigurer已弃用。

我试图找出是否有其他方法可以将 Spring-Security 与 Spring-Websocket 结合使用，但找不到相关内容。

所以，我的问题是我应该继续使用AbstractWebSocketMessageBrokerConfigurer还是有另一种方法将 Spring-Security 与 Spring-Websocket 结合起来？

这是我在测试项目中实现的示例。它似乎对你有用，但在AbstractWebSocketMessageBrokerConfigurer弃用之前不应该有 Spring 的替代方案吗？

@Configuration
public class WebSocketSecurityConfig extends AbstractWebSocketMessageBrokerConfigurer {

        protected void configureInbound(MessageSecurityMetadataSourceRegistry messages) {
                messages.simpDestMatchers("/user/*").authenticated();
        }
}

我有一个带有 Spring-Security/Session 和 Spring-Websocket 的 Spring Boot 2.2 MVC 应用程序。
它被配置为仅在经过身份验证时允许 websocket 连接。

以下是在单元测试中创建 Websocket 客户端的推荐方法。（至少我在 spring 文档中看到过这个）

private StompSession createWebsocket() throws InterruptedException, ExecutionException, TimeoutException {

    List<Transport> transports = new ArrayList<>(1);
    transports.add(new WebSocketTransport(new StandardWebSocketClient()));
    WebSocketStompClient stompClient = new WebSocketStompClient(new SockJsClient(transports));
    stompClient.setMessageConverter(new MappingJackson2MessageConverter());

    StompSession stompSession = stompClient.connect("ws://localhost:" + port + "/app",  new  StompSessionHandlerAdapter() {}).get(1, SECONDS);
    return stompSession;
}

问题

但是，这样我在日志中得到了异常，因为 Spring 重定向到登录页面，这通常是好的和需要的，因为 Websocket 请求未经身份验证。
日志确认：

2019-11-12 18:03:32.487 DEBUG 19592 --- [o-auto-1-exec-3] o.s.s.w.u.m.AndRequestMatcher            : All requestMatchers returned true
2019-11-12 18:03:32.487 DEBUG 19592 …

在我的简化案例中，我想向所有其他客户端广播 WebSocket 客户端发送的消息。该应用程序是使用响应式 websockets 和 Spring 构建的。

我的想法是使用 single Sink并且如果从客户端收到消息，则在此接收器上发出它。WebsocketSession::send只是将由此发出的事件转发Sink给连接的客户端。

@Component
class ReactiveWebSocketHandler(private val sink: Sinks.Many<Message>,
                               private val objectMapper : ObjectMapper) : WebSocketHandler {

    override fun handle(session: WebSocketSession): Mono<Void> {

        val input = session.receive()
                .doOnNext {
                    sink.emitNext(fromJson(it.payloadAsText, Message::class.java), Sinks.EmitFailureHandler.FAIL_FAST)
                }
                .then()
        val output = session.send(sink.asFlux().map { message -> session.textMessage(toJson(message)) })

        return Mono.zip(input, output).then()
    }

    fun toJson(obj : Any) : String = objectMapper.writeValueAsString(obj)

    fun <T> fromJson(json : String, clazz : Class<T>) : T{
        return objectMapper.readValue(json, clazz) …