国外建站公司,app开发公司,婺源网站建制作,手机seo关键词优化WebSocket协议定义了两种类型的消息#xff08;文本和二进制#xff09;#xff0c;但其内容未作定义。该协议定义了一种机制#xff0c;供客户端和服务器协商在WebSocket之上使用的子协议#xff08;即更高级别的消息传递协议#xff09;#xff0c;以定义各自可以发送…WebSocket协议定义了两种类型的消息文本和二进制但其内容未作定义。该协议定义了一种机制供客户端和服务器协商在WebSocket之上使用的子协议即更高级别的消息传递协议以定义各自可以发送何种消息、格式是什么、每个消息的内容等等。子协议的使用是可选的但无论如何客户端和服务器都需要就定义消息内容的一些协议达成一致。
一、概览
STOMPSimple Text Oriented Messaging Protocol最初是为脚本语言如Ruby、Python和Perl创建的用于连接到企业 message broker。它被设计用来解决常用信息传递模式的一个最小子集。STOMP可以通过任何可靠的双向流媒体网络协议使用如TCP和WebSocket。尽管STOMP是一个面向文本的协议但消息的 payload 可以是文本或二进制。
STOMP是一个基于框架的协议其框架是以HTTP为模型的。下面列出了STOMP框架的结构
COMMAND
header1:value1
header2:value2Body^
客户端可以使用 SEND 或 SUBSCRIBE 命令来发送或订阅消息以及描述消息内容和谁应该收到它的 destination header。这就实现了一个简单的发布-订阅机制你可以用它来通过 broker 向其他连接的客户端发送消息或者向服务器发送消息以请求执行某些工作。
当你使用Spring的STOMP支持时Spring WebSocket 应用程序充当客户的STOMP broker。消息被路由到 Controller 消息处理方法或简单的内存中 broker该 broker 跟踪订阅并将消息广播给订阅用户。你也可以将Spring配置为与专门的STOMP broker如RabbitMQ、ActiveMQ等合作进行消息的实际广播。在这种情况下Spring维护与 broker 的TCP连接向其转发消息并将消息从它那里传递给连接的WebSocket客户端。因此Spring web 应用可以依靠统一的基于HTTP的 security、通用验证和熟悉的编程模型来处理消息。
下面的例子显示了一个客户端订阅接收股票报价服务器可能会定期发布这些报价例如通过一个预定任务通过 SimpMessagingTemplate 向 broker 发送消息
SUBSCRIBE
id:sub-1
destination:/topic/price.stock.*^
下面的例子显示了一个客户端发送了一个交易请求服务器可以通过 MessageMapping 方法来处理
SEND
destination:/queue/trade
content-type:application/json
content-length:44{action:BUY,ticker:MMM,shares,44}^
执行后服务器可以向客户广播交易确认信息和细节。
在STOMP规范中destination 的含义是故意不透明的。它可以是任何字符串而且完全由STOMP服务器来定义它们所支持的 destination 的语义和语法。然而destination 是非常常见的它是类似路径的字符串其中 /topic/.. 意味着发布-订阅一对多/queue/ 意味着点对点一对一的消息交换。
STOMP服务器可以使用 MESSAGE 命令向所有订阅者广播信息。下面的例子显示了一个服务器向一个订阅的客户发送一个股票报价
MESSAGE
message-id:nxahklf6-1
subscription:sub-1
destination:/topic/price.stock.MMM{ticker:MMM,price:129.45}^
一个服务器不能发送未经请求的消息。所有来自服务器的消息必须是对特定客户订阅的回应而且服务器消息的 subscription header 必须与客户端 subscription 的 id header 相匹配。
前面的概述是为了提供对STOMP协议最基本的理解。我们建议阅读协议的全部 规范。
二、 好处
使用STOMP作为子协议可以让Spring框架和Spring Security提供更丰富的编程模型而不是使用原始WebSockets。关于HTTP与原始TCP的对比以及它如何让Spring MVC和其他Web框架提供丰富的功能也可以提出同样的观点。以下是一个好处清单 不需要发明一个自定义的消息传输协议和消息格式。 STOMP客户端包括Spring框架中的一个 Java客户端都是可用的。 你可以选择性地使用消息代理如RabbitMQ、ActiveMQ和其他来管理订阅和广播消息。 应用逻辑可以组织在任何数量的 Controller 实例中消息可以根据STOMP destination header 被路由到它们而不是用一个给定连接的单一 WebSocketHandler 来处理原始WebSocket消息。 你可以使用 Spring Security 来保护基于 STOMP destination 和消息类型的消息。
三、 启用 STOMP
spring-messaging 和 spring-websocket 模块中提供了基于WebSocket的STOMP的支持。一旦你有了这些依赖你就可以通过 SockJS Fallback 在WebSocket上暴露一个STOMP端点如下面的例子所示
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {Overridepublic void registerStompEndpoints(StompEndpointRegistry registry) {registry.addEndpoint(/portfolio).withSockJS(); }Overridepublic void configureMessageBroker(MessageBrokerRegistry config) {config.setApplicationDestinationPrefixes(/app); config.enableSimpleBroker(/topic, /queue); }
}/portfolio 是WebSocket或SockJS客户端需要连接以进行WebSocket握手的端点的HTTP URL。destination header 以 /app 开头的STOMP消息被路由到 Controller 类中的 MessageMapping 方法。使用内置的消息 broker 进行订阅和广播将 destination header 以 /topic 或 /queue 开头的消息路由到 broker。
下面的例子显示了前述例子的XML等效配置
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:websockethttp://www.springframework.org/schema/websocketxsi:schemaLocationhttp://www.springframework.org/schema/beanshttps://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/websockethttps://www.springframework.org/schema/websocket/spring-websocket.xsdwebsocket:message-broker application-destination-prefix/appwebsocket:stomp-endpoint path/portfoliowebsocket:sockjs//websocket:stomp-endpointwebsocket:simple-broker prefix/topic, /queue//websocket:message-broker/beans
对于内置的简单 broker/topic 和 /queue 前缀没有任何特殊含义。它们只是一个惯例用来区分发布/订阅和点对点的消息传递即许多订阅者和一个消费者。当你使用外部 broker 时请检查该 broker 的STOMP页面以了解它支持什么样的STOMP destination 和前缀。
要从浏览器连接对于SockJS你可以使用 sockjs-client。对于STOMP许多应用程序都使用了 jmesnil/stomp-websocket 库也称为stomp.js该库功能完整已在生产中使用多年但已不再维护。目前 JSteunou/webstomp-client 是该库的最积极的维护和发展的继承者。下面的示例代码是基于它的
var socket new SockJS(/spring-websocket-portfolio/portfolio);
var stompClient webstomp.over(socket);stompClient.connect({}, function(frame) {
}
另外如果你通过WebSocket没有SockJS连接你可以使用以下代码
var socket new WebSocket(/spring-websocket-portfolio/portfolio);
var stompClient Stomp.over(socket);stompClient.connect({}, function(frame) {
} 请注意前面的例子中的 stompClient 不需要指定 login 和 passcode header。即使它这样做了它们在服务器端也会被忽略或者说被覆盖。关于认证的更多信息请参见 连接到 Broker 和 认证。
更多示例代码见 使用WebSocket构建交互式wen应用程序 — 入门指南。 股票证券交易 — 示例应用。
四、 WebSocket 服务器
要配置底层的WebSocket服务器Server 配置 中的信息适用。然而对于Jetty你需要通过 StompEndpointRegistry 设置 HandshakeHandler 和 WebSocketPolicy
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {Overridepublic void registerStompEndpoints(StompEndpointRegistry registry) {registry.addEndpoint(/portfolio).setHandshakeHandler(handshakeHandler());}Beanpublic DefaultHandshakeHandler handshakeHandler() {WebSocketPolicy policy new WebSocketPolicy(WebSocketBehavior.SERVER);policy.setInputBufferSize(8192);policy.setIdleTimeout(600000);return new DefaultHandshakeHandler(new JettyRequestUpgradeStrategy(new WebSocketServerFactory(policy)));}
}五、 消息流
一旦STOMP端点被暴露Spring应用程序就成为连接客户端的STOMP broker。本节描述了服务器端的消息流。
spring-messaging 模块包含了对消息传递应用的基础支持这些支持源于 Spring Integration后来被提取并整合到Spring框架中以便在许多 Spring projects 和应用场景中更广泛地使用。下面的列表简要地描述了一些可用的消息传递messaging抽象 Message: 一个消息的简单表示包括 header 和 payload。 MessageHandler: 处理消息。 MessageChannel: 发送消息使生产者和消费者之间实现松散耦合。 SubscribableChannel: 带有 MessageHandler 订阅者的 MessageChannel。 ExecutorSubscribableChannel: SubscribableChannel使用一个 Executor 来传递消息。
Java 配置即 EnableWebSocketMessageBroker和 XML 命名空间配置即 websocket:message-broker都使用前面的组件来组装一个消息工作流。下图显示了启用简单内置消息 broker 时使用的组件 前面的图显示了三个信息通道message channel clientInboundChannel: 用于传递从WebSocket客户端收到的消息。 clientOutboundChannel: 用于向WebSocket客户端发送服务器信息。 brokerChannel: 用于从服务器端的应用程序代码中向消息 broker 发送消息。
下图显示了当外部 broker如 RabbitMQ被配置为管理订阅和广播消息时使用的组件 前面两张图的主要区别是使用 “broker relay 中继”通过TCP将消息向上传递到外部STOMP broker并从 broker 向下传递消息到订阅的客户。
当从WebSocket连接中接收到消息时它们会被解码为STOMP帧变成Spring Message 表示并被发送到 clientInboundChannel 进行进一步处理。例如destination header 以 /app 开头的 STOMP 消息可以被路由到注解 controller 中的 MessageMapping 方法而 /topic 和 /queue 消息可以直接被路由到消息 broker。
一个处理来自客户端的STOMP消息的注解的 Controller 可以通过 brokerChannel 向消息代理发送消息而代理则通过 clientOutboundChannel 将消息广播给匹配的订阅者。同样的 controller 也可以在响应HTTP请求时做同样的事情所以客户端可以执行一个HTTP POST然后一个 PostMapping 方法可以发送一个消息给消息 broker以广播给订阅的客户端。
我们可以通过一个简单的例子来追踪这个流程。考虑下面这个例子它设置了一个服务器
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {Overridepublic void registerStompEndpoints(StompEndpointRegistry registry) {registry.addEndpoint(/portfolio);}Overridepublic void configureMessageBroker(MessageBrokerRegistry registry) {registry.setApplicationDestinationPrefixes(/app);registry.enableSimpleBroker(/topic);}
}Controller
public class GreetingController {MessageMapping(/greeting)public String handle(String greeting) {return [ getTimestamp() : greeting;}
}前面的例子支持以下流程 客户端连接到 http://localhost:8080/portfolio一旦建立WebSocket连接STOMP帧就开始在上面流动。 客户端发送一个 SUBSCRIBE 帧destination header 为 /topic/greeting。一旦收到并解码该消息被发送到 clientInboundChannel然后被路由到消息 broker该 broker 存储客户端的订阅。 客户端发送一个 SEND 帧到 /app/greeting。/app 前缀有助于将其路由到有注解的 controller 。在 /app 前缀被剥离后剩余的 /greeting 部分的 destination 被映射到 GreetingController 的 MessageMapping 方法。 从 GreetingController 返回的值被转化为Spring Message其 payload 基于返回值和默认的 destination header /topic/greeting从输入 destination 派生用 /topic 代替 /app。产生的消息被发送到 brokerChannel 并由消息 broker 处理。 消息 broker 找到所有匹配的订阅者并通过客户端 OutboundChannel 向每个订阅者发送一个 MESSAGE 帧从那里消息被编码为STOMP帧并在WebSocket连接上发送。
下一节将提供关于注解方法的更多细节包括支持的参数和返回值的种类。
六、 注解式Controller
应用程序可以使用注解的 Controller 类来处理来自客户端的消息。这样的类可以声明 MessageMapping、SubscribeMapping 和 ExceptionHandler 方法如以下主题所述 MessageMapping SubscribeMapping MessageExceptionHandler
MessageMapping
你可以使用 MessageMapping 来注解那些根据 destination 路由消息的方法。它在方法层面和类型层面都被支持。在类型层面上MessageMapping 被用来表达 controller 中所有方法的共享映射。
默认情况下映射值是Ant风格的路径模式例如 /thing*/thing/**包括对模板变量的支持例如/thing/{id}。这些值可以通过 DestinationVariable 方法参数进行引用。应用程序也可以切换到点状分隔的目标约定来进行映射正如在 使用点作为分隔符 中解释的那样。
支持的方法参数
下表描述了方法的参数
方法参数说明 Message 为了获得完整的信息。 MessageHeaders 用于访问 Message 中的 header。 MessageHeaderAccessor, SimpMessageHeaderAccessor, and StompHeaderAccessor 用于通过类型化的访问器方法访问 header。 Payload 用于访问消息的 payload由配置的 MessageConverter 转换例如从JSON。 这个注解的存在不是必须的因为默认情况下如果没有其他参数被匹配它就会被假定。 你可以用 jakarta.validation.Valid 或Spring的 Validated 来注解 payload 参数以使 payload 参数被自动验证。 Header 用于访问一个特定的 header 值—如果有必要同时使用 org.springframework.core.convert.Converter.Converter 进行类型转换。 Headers 用于访问消息中的所有 header。这个参数必须是可以分配给 java.util.Map。 DestinationVariable 用于访问从消息 destination 提取的模板变量。必要时值被转换为声明的方法参数类型。 java.security.Principal 反映在WebSocket HTTP握手时登录的用户。
返回值
默认情况下MessageMapping 方法的返回值通过匹配的 MessageConverter 被序列化为一个 payload并作为一个 Message 发送到 brokerChannel从那里被广播给订阅者。出站消息的 destination 与入站消息的 destination 相同但前缀为 /topic。
你可以使用 SendTo 和 SendToUser 注解来定制输出消息的 destination。SendTo 是用来定制目标 destination 或指定多个 destination 的。 SendToUser 用来指导输出消息只给与输入消息相关的用户。参见 User Destination。
你可以在同一个方法上同时使用 SendTo 和 SendToUser而且在类的层面上也支持这两种注解在这种情况下它们作为类中方法的默认值。然而请记住任何方法级的 SendTo 或 SendToUser 注解都会覆盖类级的任何此类注解。
消息可以被异步处理MessageMapping 方法可以返回 ListenableFuture、 CompletableFuture 或 CompletionStage。
请注意SendTo 和 SendToUser 只是一种便利相当于使用 SimpMessagingTemplate 来发送消息。如果有必要对于更高级的场景 MessageMapping 方法可以直接使用 SimpMessagingTemplate。这可以代替返回值也可以在返回值之外进行。参见 发送消息。
SubscribeMapping
SubscribeMapping 与 MessageMapping 类似但只将映射范围缩小到订阅信息。它支持与 MessageMapping 相同的 方法参数。然而对于返回值默认情况下消息被直接发送到客户端通过 clientOutboundChannel对订阅的响应而不是发送到 broker通过 brokerChannel作为广播给匹配的订阅。添加 SendTo 或 SendToUser 会重写这一行为并代替发送至 broker。
这在什么时候是有用的假设 broker 被映射到 /topic 和 /queue而应用 controller 被映射到 /app。在这种设置下broker 存储了所有对 /topic 和 /queue 的订阅这些订阅是为了重复广播而应用程序不需要参与。客户端也可以订阅一些 /app 的 destinationcontroller 可以在不涉及 broker 的情况下返回一个值而不需要再次存储或使用该订阅实际上是一个一次性的请求-回复 exchange。这方面的一个用例是在启动时用初始数据填充一个用户界面。
这在什么时候没有用不要试图将 broker 和 controller 映射到同一个目标前缀除非你想让两者独立处理消息包括订阅因为某些原因。入站消息是平行处理的。不能保证一个 broker 或一个 controller 先处理一个给定的消息。如果目标是在订阅被存储并准备好进行广播时得到通知如果服务器支持的话客户端应该要求得到一个 receipt 简单的 broker 不支持。例如使用Java STOMP client你可以做以下事情来添加一个 receipt
Autowired
private TaskScheduler messageBrokerTaskScheduler;// During initialization..
stompClient.setTaskScheduler(this.messageBrokerTaskScheduler);// When subscribing..
StompHeaders headers new StompHeaders();
headers.setDestination(/topic/...);
headers.setReceipt(r1);
FrameHandler handler ...;
stompSession.subscribe(headers, handler).addReceiptTask(receiptHeaders - {// Subscription ready...
});一个服务器端的选择是在 brokerChannel 上 注册 一个 ExecutorChannelInterceptor并实现 afterMessageHandled 方法该方法在消息包括订阅被处理后被调用。
MessageExceptionHandler
一个应用程序可以使用 MessageExceptionHandler 方法来处理来自 MessageMapping 方法的异常。你可以在注解本身中声明异常如果你想获得对异常实例的访问也可以通过方法参数来声明。下面的例子通过一个方法参数声明了一个异常
Controller
public class MyController {// ...MessageExceptionHandlerpublic ApplicationError handleException(MyException exception) {// ...return appError;}
}MessageExceptionHandler 方法支持灵活的方法签名支持与 MessageMapping 方法相同的方法参数类型和返回值。
通常情况下MessageExceptionHandler 方法适用于它们所声明的 Controller 类或类层次结构。如果你想让这些方法在全局范围内应用跨控制器你可以在一个标有 ControllerAdvice 的类中声明它们。这与Spring MVC中的类似 支持相当。
七、 发送消息
如果你想从应用程序的任何部分向连接的客户端发送消息怎么办任何应用程序组件都可以向 brokerChannel 发送消息。最简单的方法是注入一个 SimpMessagingTemplate 并使用它来发送消息。通常情况下你会按类型注入它如下例所示
Controller
public class GreetingController {private SimpMessagingTemplate template;Autowiredpublic GreetingController(SimpMessagingTemplate template) {this.template template;}RequestMapping(path/greetings, methodPOST)public void greet(String greeting) {String text [ getTimestamp() ]: greeting;this.template.convertAndSend(/topic/greetings, text);}}然而你也可以通过它的名字brokerMessagingTemplate来限定它如果存在另一个相同类型的bean。
八、Simple Broker
内置的简单 message broker 处理来自客户端的订阅请求将其存储在内存中并将消息广播给有匹配目的地的连接客户端。broker 支持类似路径的 destination包括对 Ant 风格的 destination 模式的订阅。
应用程序也可以使用点分割的而不是斜线分割的destination。参见 使用点作为分隔符。
如果配置了一个任务调度器task schedulersimple broker 就支持 STOMP 心跳。要配置一个调度器你可以声明你自己的 TaskScheduler Bean并通过 MessageBrokerRegistry 来设置它。或者你可以使用在内置 WebSocket 配置中自动声明的那个但是你将’需要 Lazy 以避免内置 WebSocket 配置和你的 WebSocketMessageBrokerConfigurer 之间的循环。例如
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {private TaskScheduler messageBrokerTaskScheduler;Autowiredpublic void setMessageBrokerTaskScheduler(Lazy TaskScheduler taskScheduler) {this.messageBrokerTaskScheduler taskScheduler;}Overridepublic void configureMessageBroker(MessageBrokerRegistry registry) {registry.enableSimpleBroker(/queue/, /topic/).setHeartbeatValue(new long[] {10000, 20000}).setTaskScheduler(this.messageBrokerTaskScheduler);// ...}
}九、 外部 Broker
简单的 broker 很适合入门但它只支持STOMP命令的一个子集它不支持acks、receipts和其他一些功能依赖于一个简单的消息发送循环并且不适合集群化。作为一种选择你可以升级你的应用程序以使用全功能的消息代理。
请参阅你所选择的消息代理如 RabbitMQ、 ActiveMQ 等的 STOMP 文档安装该代理并在启用 STOMP 支持的情况下运行它。然后你可以在Spring配置中启用STOMP broker 中继而不是简单的 broker。
下面的配置示例启用了一个全功能的 broker
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {Overridepublic void registerStompEndpoints(StompEndpointRegistry registry) {registry.addEndpoint(/portfolio).withSockJS();}Overridepublic void configureMessageBroker(MessageBrokerRegistry registry) {registry.enableStompBrokerRelay(/topic, /queue);registry.setApplicationDestinationPrefixes(/app);}}下面的例子显示了前述例子的XML配置等效
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:websockethttp://www.springframework.org/schema/websocketxsi:schemaLocationhttp://www.springframework.org/schema/beanshttps://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/websockethttps://www.springframework.org/schema/websocket/spring-websocket.xsdwebsocket:message-broker application-destination-prefix/appwebsocket:stomp-endpoint path/portfolio /websocket:sockjs//websocket:stomp-endpointwebsocket:stomp-broker-relay prefix/topic,/queue //websocket:message-broker/beans
前面配置中的STOMP broker 中继器是一个Spring MessageHandler它通过将消息转发给外部 message broker 来处理消息。为此它建立了与 broker 的TCP连接将所有消息转发给 broker 然后将从 broker 那里收到的所有消息通过WebSocket会话转发给客户。从本质上讲它充当了一个 中转站在两个方向上转发消息。
将 io.projectreactor.netty:reactor-netty 和 io.netty:netty-all 依赖项添加到你的项目中用于 TCP 连接管理。
此外应用组件如HTTP请求处理方法、业务服务等也可以向 broker 中继发送消息如 发送消息 中所述以向订阅的WebSocket客户端广播消息。
实际上broker 中继实现了稳健和可扩展的消息广播。
十、连接到 Broker
STOMP broker 中转站保持着一个与 broker 的单一 系统 TCP连接。这个连接只用于从服务器端应用程序发出的消息不用于接收消息。你可以为这个连接配置STOMP凭证即STOMP框架 login 和 passcode header。这在XML命名空间和Java配置中都被暴露为 systemLogin 和 systemPasscode 属性默认值为 guest 和 guest。
STOMP broker 中继器还为每个连接的 WebSocket 客户端创建一个单独的TCP连接。你可以配置 STOMP 凭证这些凭证用于代表客户创建的所有TCP连接。这在XML命名空间和Java配置中都被暴露为 clientLogin 和 clientPasscode 属性默认值为 guest 和 guest。
STOMP broker 中继总是在它代表客户转发给 broker 的每个 CONNECT 帧上设置 login 和 passcode header。因此WebSocket客户端不需要设置这些 header 信息。它们会被忽略。正如 认证 部分所解释的WebSocket客户端应依靠HTTP认证来保护WebSocket端点并建立客户端身份。
STOMP broker 中继也通过 系统 TCP连接向 message broker 发送和接收心跳。你可以配置发送和接收心跳的时间间隔默认为10秒。如果与 broker 的连接丢失broker 中继会继续尝试重新连接每5秒一次直到成功。
任何Spring Bean都可以实现 ApplicationListenerBrokerAvailabilityEvent 来接收与 broker 的 系统 连接丢失和重新建立时的通知。例如当没有活跃的 系统 连接时一个广播股票报价的股票报价服务可以停止尝试发送消息。
默认情况下STOMP broker 中继器总是连接到同一主机和端口并在失去连接时根据需要重新连接。如果你希望提供多个地址在每次尝试连接时你可以配置一个地址 supplier而不是一个固定的主机和端口。下面的例子显示了如何做到这一点
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {// ...Overridepublic void configureMessageBroker(MessageBrokerRegistry registry) {registry.enableStompBrokerRelay(/queue/, /topic/).setTcpClient(createTcpClient());registry.setApplicationDestinationPrefixes(/app);}private ReactorNettyTcpClientbyte[] createTcpClient() {return new ReactorNettyTcpClient(client - client.addressSupplier(() - ... ),new StompReactorNettyCodec());}
}你也可以用 virtualHost 属性来配置STOMP broker 中继。这个属性的值被设置为每个 CONNECT 帧的 host header可能很有用例如在云环境中建立TCP连接的实际主机与提供基于云的STOMP服务的主机不同。
十一、 使用点作为分隔符
当消息被路由到 MessageMapping 方法时它们会与 AntPathMatcher 匹配。默认情况下pattern 被期望使用斜线/作为分隔符。这在web应用中是一个很好的惯例与HTTP URLs类似。然而如果你更习惯于信息传递的惯例你可以切换到使用点.作为分隔符。
下面的例子显示了如何在Java配置中这样做
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {// ...Overridepublic void configureMessageBroker(MessageBrokerRegistry registry) {registry.setPathMatcher(new AntPathMatcher(.));registry.enableStompBrokerRelay(/queue, /topic);registry.setApplicationDestinationPrefixes(/app);}
}下面的例子显示了前述例子的XML等效配置
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:websockethttp://www.springframework.org/schema/websocketxsi:schemaLocationhttp://www.springframework.org/schema/beanshttps://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/websockethttps://www.springframework.org/schema/websocket/spring-websocket.xsdwebsocket:message-broker application-destination-prefix/app path-matcherpathMatcherwebsocket:stomp-endpoint path/stomp/websocket:stomp-broker-relay prefix/topic,/queue //websocket:message-brokerbean idpathMatcher classorg.springframework.util.AntPathMatcherconstructor-arg index0 value.//bean/beans
之后controller 可以在 MessageMapping 方法中使用点.作为分隔符如下图所示
Controller
MessageMapping(red)
public class RedController {MessageMapping(blue.{green})public void handleGreen(DestinationVariable String green) {// ...}
}客户端现在可以向 /app/red.blue.green123 发送消息。
在前面的例子中我们没有改变 “broker relay” 上的前缀因为这些完全取决于外部 message broker。请看你使用的 broker 的STOMP文档页面看看它对 destination header 支持什么约定。
另一方面“simple broker” 确实依赖于配置的 PathMatcher所以如果你切换分隔符这种变化也适用于 broker 和 broker 将消息中的 destination 与订阅中的 pattern 相匹配的方式。
十二、认证
每个基于WebSocket的STOMP 消息会话都从一个HTTP请求开始。这可以是一个升级到WebSocket的请求即WebSocket握手或者在SockJS回退的情况下是一系列的SockJS HTTP传输请求。
许多Web应用程序已经有了认证和授权以确保HTTP请求的安全。通常情况下用户通过Spring Security使用某种机制进行认证如登录页面、HTTP基本认证或其他方式。认证用户的 security context 被保存在HTTP会话中并与同一基于cookie的会话中的后续请求相关。
因此对于WebSocket握手或SockJS的HTTP传输请求通常已经有一个可通过 HttpServletRequest#getUserPrincipal() 访问的认证用户。Spring会自动将该用户与为其创建的WebSocket或SockJS会话联系起来随后通过 user header 通过该会话传输的所有STOMP消息联系起来。
简而言之一个典型的web应用程序除了已经做的安全工作外不需要做任何其他事情。用户在HTTP请求层面通过 security context 进行认证该 security context 通过基于cookie的HTTP会话然后与为该用户创建的WebSocket或SockJS会话相关联进行维护并导致在流经应用程序的每个 Message 上印上一个 user header。
STOMP协议在 CONNECT 帧上确实有 login 和 passcode header。这些头信息最初是为通过TCP的STOMP而设计的并且是需要的。然而对于通过WebSocket的STOMP默认情况下Spring忽略了STOMP协议级别的 authentication header并假定用户已经在HTTP传输级别进行了认证。我们期望WebSocket或SockJS会话包含认证的用户。
十三、 Token 认证
Spring Security OAuth 提供了对基于令牌的安全支持包括 JSON Web TokenJWT。你可以在Web应用程序中使用它作为认证机制包括上一节所述的基于WebSocket的STOMP交互也就是通过基于cookie的会话来维护身份。
同时基于cookie的会话并不总是最合适的例如在不维护服务器端会话的应用中或者在移动应用中通常使用 header 信息进行认证。
WebSocket协议RFC 6455 没有规定服务器在WebSocket握手过程中对客户端进行认证的任何特定方式。然而在实践中浏览器客户端只能使用标准 authentication header即 basic HTTP authentication或cookies不能例如提供自定义 header。同样地SockJS的JavaScript客户端也没有提供与SockJS传输请求一起发送HTTP header 的方法。见 sockjs-client issue 196。相反它确实允许发送查询参数你可以用它来发送 token但这也有自己的缺点例如token 可能会无意中与服务器日志中的URL一起被记录。
前面的限制是针对基于浏览器的客户端不适用于基于Spring Java 的 STOMP client该客户端确实支持通过WebSocket和SockJS请求发送 header 信息。
因此希望避免使用cookies的应用程序在HTTP协议层面上可能没有任何好的替代认证。与其使用cookies他们可能更喜欢在STOMP消息协议层面用 header 进行认证。这样做需要两个简单的步骤 在连接时使用STOMP客户端来传递 authentication header 信息。 用一个 ChannelInterceptor 来处理 authentication header。
下一个例子使用服务器端配置来注册一个自定义认证拦截器。请注意拦截器只需要在 CONNECT Message 上进行认证和设置 user header。Spring 会记录并保存认证的用户并将其与同一会话中的后续STOMP消息相关联。下面的例子展示了如何注册一个自定义认证拦截器
Configuration
EnableWebSocketMessageBroker
public class MyConfig implements WebSocketMessageBrokerConfigurer {Overridepublic void configureClientInboundChannel(ChannelRegistration registration) {registration.interceptors(new ChannelInterceptor() {Overridepublic Message? preSend(Message? message, MessageChannel channel) {StompHeaderAccessor accessor MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);if (StompCommand.CONNECT.equals(accessor.getCommand())) {Authentication user ... ; // access authentication header(s)accessor.setUser(user);}return message;}});}
}此外请注意当你使用Spring Security的消息授权时目前你需要确保认证 ChannelInterceptor 配置的顺序在Spring Security的前面。要做到这一点最好是在自己的 WebSocketMessageBrokerConfigurer 实现中声明自定义拦截器并标明 Order(Ordered.HIGHEST_PRECEDENCE 99)。
十四、 授权
Spring Security提供了 WebSocket 子协议授权它使用 ChannelInterceptor 来根据消息中的 user header 授权。此外Spring Session还提供了 WebSocket integration确保用户的HTTP会话不会过期而WebSocket会话仍在活动。
十五、 User Destination
应用程序可以发送针对特定用户的消息Spring的STOMP支持为此 destination 识别以 /user/ 为前缀的 destination。例如一个客户端可能会订阅 /user/queue/position-updates destination。UserDestinationMessageHandler 处理这个 destination并将其转换为用户会话的唯一 destination如 /queue/position-updates-user123。这为订阅一个通用命名的 destination 提供了便利同时确保不会与订阅同一 destination 的其他用户发生冲突以便每个用户都能收到独特的股票位置更新。
当使用用户 destination 时重要的是要配置 broker 和应用程序的 destination 前缀如 启用 STOMP, 所示否则 broker 将处理 /user 前缀的消息而这些消息只应该由 UserDestinationMessageHandler 处理。
在发送方面消息可以被发送到一个 destination如 /user/{username}/queue/position-updates这又被 UserDestinationMessageHandler 翻译成一个或多个destination每个与用户相关的会话都有一个。这让应用程序中的任何组件都可以发送针对特定用户的消息而不一定要知道他们的名字和通用destination。这也是通过一个注解和一个消息模板来支持的。
一个消息处理方法可以通过 SendToUser 注解将消息发送给与被处理的消息相关的用户在类级上也支持共享一个共同的 destination如下例所示
Controller
public class PortfolioController {MessageMapping(/trade)SendToUser(/queue/position-updates)public TradeResult executeTrade(Trade trade, Principal principal) {// ...return tradeResult;}
}如果用户有一个以上的会话默认情况下所有订阅给定 destination 的会话都是目标。然而有时可能需要只针对发送被处理消息的会话。你可以通过将 broadcast 属性设置为 false 来做到这一点正如下面的例子所示
Controller
public class MyController {MessageMapping(/action)public void handleAction() throws Exception{// raise MyBusinessException here}MessageExceptionHandlerSendToUser(destinations/queue/errors, broadcastfalse)public ApplicationError handleException(MyBusinessException exception) {// ...return appError;}
}虽然用户 destination 通常意味着有一个经过认证的用户但这并不是严格的要求。不与认证用户相关联的WebSocket会话可以订阅用户 destination。在这种情况下 SendToUser 注解的行为与 broadcastfalse 时完全相同即只针对发送被处理消息的会话。
你可以从任何应用组件向用户 destination 发送消息例如通过注入由Java配置或XML命名空间创建的 SimpMessagingTemplate。(如果需要用 Qualifier 来限定bean的名字是 brokerMessagingTemplate)。下面的例子显示了如何做到这一点
Service
public class TradeServiceImpl implements TradeService {private final SimpMessagingTemplate messagingTemplate;Autowiredpublic TradeServiceImpl(SimpMessagingTemplate messagingTemplate) {this.messagingTemplate messagingTemplate;}// ...public void afterTradeExecuted(Trade trade) {this.messagingTemplate.convertAndSendToUser(trade.getUserName(), /queue/position-updates, trade.getResult());}
}当你将用户 destination 与外部消息 broker 一起使用时你应该查看关于如何管理非活动队列的 broker 文档以便在用户会话结束时所有独特的用户队列都被删除。例如当你使用诸如 /exchange/amq.direct/position-updates 等 destination 时RabbitMQ 会创建自动删除队列。因此在这种情况下客户端可以订阅到 /user/exchange/amq.direct/position-updates。同样地ActiveMQ也有清除非活动 destination 的 配置选项。
在一个多应用服务器的情况下一个用户 destination 可能会因为用户连接到一个不同的服务器而保持未解析。在这种情况下你可以配置一个 destination 来广播未解析的消息以便其他服务器有机会尝试。这可以通过Java配置中 MessageBrokerRegistry 的 userDestinationBroadcast 属性和XML中 message-broker 元素的 user-destination-broadcast 属性完成。
十六、 消息的顺序
来自 broker 的消息被发布到客户端 OutboundChannel从那里被写入WebSocket会话。由于该通道由 ThreadPoolExecutor 支持消息在不同的线程中被处理而客户端收到的结果序列可能与发布的确切顺序不一致。
如果这是一个问题请启用 setPreservePublishOrder 标志如下例所示
Configuration
EnableWebSocketMessageBroker
public class MyConfig implements WebSocketMessageBrokerConfigurer {Overrideprotected void configureMessageBroker(MessageBrokerRegistry registry) {// ...registry.setPreservePublishOrder(true);}}下面的例子显示了前述例子的XML等效配置
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:websockethttp://www.springframework.org/schema/websocketxsi:schemaLocationhttp://www.springframework.org/schema/beanshttps://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/websockethttps://www.springframework.org/schema/websocket/spring-websocket.xsdwebsocket:message-broker preserve-publish-ordertrue!-- ... --/websocket:message-broker/beans 当该标志被设置时同一客户端会话中的消息会被一个一个地发布到 clientOutboundChannel 上这样就保证了发布的顺序。请注意这将产生一个小的性能开销所以你应该只在需要时才启用它。
十七、事件
几个 ApplicationContext 事件event被发布并可以通过实现Spring的 ApplicationListener 接口来接收 BrokerAvailabilityEvent: 指示 broker 何时可用或不可用。虽然 “simple” broker 在启动时立即可用并在应用程序运行时保持可用但STOMP “broker relay” 可能会失去与全功能 broker 的连接例如如果 broker 被重新启动。broker 中继有重新连接的逻辑当它回来时重新建立与 broker 的 系统 连接。因此只要状态从连接变为断开反之亦然就会发布这个事件。使用 SimpMessagingTemplate 的组件应该订阅这个事件并避免在 broker 不可用时发送消息。在任何情况下他们应该准备好在发送消息时处理 MessageDeliveryException。 SessionConnectEvent: 当收到一个新的 STOMP CONNECT 时发布表示一个新的客户端会话的开始。该事件包含代表连接的消息包括会话ID、用户信息如果有的话以及客户端发送的任何自定义头信息。这对于跟踪客户端会话是很有用的。订阅此事件的组件可以用 SimpMessageHeaderAccessor 或 StompMessageHeaderAccessor 来包装所包含的消息。 SessionConnectedEvent: 在 SessionConnectEvent 发生后不久当 broker 发送一个 STOMP CONNECTED 帧作为对 CONNECT 的响应时发布。在这一点上STOMP会话可被视为完全建立。 SessionSubscribeEvent: 当收到一个新的 STOMP SUBSCRIBE 时发布。 SessionUnsubscribeEvent: 当收到一个新的 STOMP UNSUBSCRIBE 时发布。 SessionDisconnectEvent: 在STOMP会话结束时发布。DISCONNECT 可能是由客户端发送的也可能是在WebSocket会话关闭时自动生成的。在某些情况下该事件会在每个会话中发布一次以上。对于多个断开事件组件应该是幂等的。
当你使用一个全功能的 broker 时如果 broker 暂时不可用STOMP 的 “broker relay” 会自动重新连接 系统 的连接。然而客户端连接不会自动重新连接。假设心跳被启用客户端通常会在10秒内注意到 broker 没有回应。客户端需要实现他们自己的重新连接逻辑。
十入、 拦截
事件 为STOMP连接的生命周期提供通知但不是为每个客户端消息提供通知。应用程序也可以注册一个 ChannelInterceptor 来拦截任何消息和处理链中的任何部分。下面的例子显示了如何拦截来自客户端的入站消息
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {Overridepublic void configureClientInboundChannel(ChannelRegistration registration) {registration.interceptors(new MyChannelInterceptor());}
}自定义 ChannelInterceptor 可以使用 StompHeaderAccessor 或 SimpMessageHeaderAccessor 来访问消息的信息如下图所示
public class MyChannelInterceptor implements ChannelInterceptor {Overridepublic Message? preSend(Message? message, MessageChannel channel) {StompHeaderAccessor accessor StompHeaderAccessor.wrap(message);StompCommand command accessor.getStompCommand();// ...return message;}
}应用程序也可以实现 ExecutorChannelInterceptor它是 ChannelInterceptor 的一个子接口在处理消息的线程中具有回调功能。 ChannelInterceptor 对于发送到 channel 的每个消息都会被调用一次而 ExecutorChannelInterceptor 则在订阅通道消息的每个 MessageHandler 的线程中提供钩子hook。
注意与前面描述的 SessionDisconnectEvent 一样DISCONNECT 消息可以来自客户端也可以在WebSocket会话关闭时自动生成。在某些情况下拦截器可以为每个会话拦截该消息一次以上。对于多个断开连接事件组件应该是幂等的。
十九、 STOMP 客户端
Spring提供了一个基于 WebSocket 的 STOMP 客户端和一个基于 TCP 的 STOMP 客户端。
首先你可以创建并配置 WebSocketStompClient如下例所示
WebSocketClient webSocketClient new StandardWebSocketClient();
WebSocketStompClient stompClient new WebSocketStompClient(webSocketClient);
stompClient.setMessageConverter(new StringMessageConverter());
stompClient.setTaskScheduler(taskScheduler); // for heartbeats在前面的例子中你可以用 SockJsClient 取代 StandardWebSocketClient因为它也是 WebSocketClient 的一个实现。SockJsClient 可以使用WebSocket或基于HTTP的传输作为后退。更多详情请参见 SockJsClient。
接下来你可以建立一个连接并为 STOMP 会话提供一个 handler如下例所示
String url ws://127.0.0.1:8080/endpoint;
StompSessionHandler sessionHandler new MyStompSessionHandler();
stompClient.connect(url, sessionHandler);当会话可以使用时handler 会被通知如下例所示
public class MyStompSessionHandler extends StompSessionHandlerAdapter {Overridepublic void afterConnected(StompSession session, StompHeaders connectedHeaders) {// ...}
}一旦会话建立任何 payload 都可以被发送并通过配置的 MessageConverter 进行序列化如下例所示
session.send(/topic/something, payload);你也可以对 destination 进行订阅。subscribe 方法需要一个 handler 来处理订阅上的消息并返回一个 Subscription handle你可以用它来取消订阅。对于每个收到的消息handler 可以指定 payload 应该被反序列化的目标 Object 类型如下面的例子所示
session.subscribe(/topic/something, new StompFrameHandler() {Overridepublic Type getPayloadType(StompHeaders headers) {return String.class;}Overridepublic void handleFrame(StompHeaders headers, Object payload) {// ...}});为了启用STOMP心跳你可以用一个 TaskScheduler 来配置 WebSocketStompClient并可选择自定义心跳间隔10 秒写不活动会导致发送心跳10 秒读不活动会关闭连接。
WebSocketStompClient 只在不活动的情况下发送心跳即没有其他消息被发送时。当使用外部 broker时这可能是一个挑战因为具有非 broker destination 的消息代表活动但实际上并没有转发给 broker。在这种情况下你可以在初始化 外部 Broker 时配置一个 TaskScheduler以确保在只有非 broker destination 的消息被发送时心跳也被转发到 broker。
当你使用 WebSocketStompClient 进行性能测试以模拟来自同一台机器的数千个客户端时请考虑关闭心跳因为每个连接都会安排自己的心跳任务而这对于在同一台机器上运行的大量客户端来说并不优化。
STOMP协议还支持 receipt客户端必须添加一个 receipt 头服务器在处理完发送或订阅后会用一个 RECEIPT 帧来回应。为了支持这一点StompSession 提供了 setAutoReceipt(boolean) 功能使每一个后续的发送或订阅事件都会添加一个 receipt 头。另外你也可以手动添加一个 receipt 头到 StompHeaders 中。发送和订阅都会返回一个 Receiptable 的实例你可以用它来注册接收成功和失败的回调。对于这个功能你必须用一个 TaskScheduler 和 receipt 过期前的时间量默认为15秒来配置客户端。
请注意StompSessionHandler 本身就是一个 StompFrameHandler这让它除了处理信息处理中的异常的 handleException 回调和处理包括 ConnectionLostException 在内的传输级错误的 handleTransportError 之外还可以处理ERROR帧。
二十、WebSocket Scope
每个WebSocket会话都有一个 attributes map。该 map 作为 header 附在入站的客户端消息上可以从 controller 方法中访问如下例所示
Controller
public class MyController {MessageMapping(/action)public void handle(SimpMessageHeaderAccessor headerAccessor) {MapString, Object attrs headerAccessor.getSessionAttributes();// ...}
}你可以在 websocket scope 内声明一个Spring管理的Bean。你可以将 WebSocket scope 的 Bean 注入 controller 和在 clientInboundChannel 上注册的任何通道拦截器中。这些通常是 singleton生命周期比任何单独的 WebSocket 会话长。因此你需要为 WebSocket scope 的 Bean 使用 scope proxy 模式如下例所示
Component
Scope(scopeName websocket, proxyMode ScopedProxyMode.TARGET_CLASS)
public class MyBean {PostConstructpublic void init() {// Invoked after dependencies injected}// ...PreDestroypublic void destroy() {// Invoked when the WebSocket session ends}
}Controller
public class MyController {private final MyBean myBean;Autowiredpublic MyController(MyBean myBean) {this.myBean myBean;}MessageMapping(/action)public void handle() {// this.myBean from the current WebSocket session}
}与任何自定义 scope 一样Spring在第一次从 controller 访问 MyBean 时初始化一个新的 MyBean 实例并将该实例存储在WebSocket会话属性中。随后会返回相同的实例直到会话结束。如前面的例子所示WebSocket scope 的Bean有所有Spring生命周期方法被调用。
二十一、 性能
谈到性能没有银弹。许多因素都会影响它包括消息的大小和数量应用程序方法是否执行需要阻塞的工作以及外部因素如网络速度和其他问题。本节的目的是提供一个可用配置选项的概述以及关于如何推理扩展的一些想法。
在一个消息传递的应用程序中消息是通过通道传递的用于由线程池支持的异步执行。配置这样一个应用程序需要对通道和消息流有很好的了解。因此我们建议回顾一下 消息流。
最明显的地方是配置支持 clientInboundChannel 和 clientOutboundChannel 的线程池。默认情况下两者都被配置为可用处理器数量的两倍。
如果注解方法中的消息处理主要是由CPU约束的那么 clientInboundChannel 的线程数量应该保持与处理器的数量接近。如果它们所做的工作更多的是IO-bound需要在数据库或其他外部系统上阻塞或等待那么线程池的大小可能需要增加。 ThreadPoolExecutor 有三个重要的属性核心线程池大小最大线程池大小以及用于存储没有可用线程的任务的队列容量。 一个常见的混淆点是配置核心池大小例如10和最大池大小例如20的结果是线程池有10到20个线程。事实上如果容量保持在 Integer.MAX_VALUE 的默认值线程池的增加永远不会超过核心池的大小因为所有额外的任务都是排队的。 请参阅 ThreadPoolExecutor 的javadoc了解这些属性如何工作并理解各种队列策略。
在 clientOutboundChannel 方面它是向WebSocket客户端发送消息的全部内容。如果客户处于告速网络上线程数应保持接近可用处理器的数量。如果他们速度慢或带宽低他们需要更长的时间来消费消息给线程池带来负担。因此增加线程池的大小成为必要。
虽然 clientInboundChannel 的工作量是可以预测的—毕竟它是基于应用程序所做的事情—但如何配置 clientOutboundChannel 则比较困难因为它是基于应用程序无法控制的因素。出于这个原因有两个额外的属性与消息的发送有关sendTimeLimit 和 sendBufferSizeLimit。你可以使用这些方法来配置在向客户端发送消息时允许发送多长时间以及可以缓冲多少数据。
一般的想法是在任何时候只有一个线程可以用来向客户端发送。同时所有额外的消息都会被缓冲你可以使用这些属性来决定允许发送一个消息需要多长时间以及在此期间有多少数据可以被缓冲。关于其他重要的细节请参见javadoc和XML schema 的文档。
下面的例子显示了一个可能的配置
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {Overridepublic void configureWebSocketTransport(WebSocketTransportRegistration registration) {registration.setSendTimeLimit(15 * 1000).setSendBufferSizeLimit(512 * 1024);}// ...}下面的例子显示了前述例子的XML等效配置
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:websockethttp://www.springframework.org/schema/websocketxsi:schemaLocationhttp://www.springframework.org/schema/beanshttps://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/websockethttps://www.springframework.org/schema/websocket/spring-websocket.xsdwebsocket:message-brokerwebsocket:transport send-timeout15000 send-buffer-size524288 /!-- ... --/websocket:message-broker/beans 你也可以使用前面显示的WebSocket传输配置来配置传入STOMP消息的最大允许大小。理论上WebSocket消息的大小几乎没有限制。在实践中WebSocket服务器会施加限制—例如Tomcat上的8K和Jetty上的64K。出于这个原因STOMP客户端如JavaScript webstomp-client 和其他客户端将较大的STOMP消息以16K的边界分割并作为多个WebSocket消息发送这需要服务器进行缓冲和重新组合。
Spring的基于 WebSocket 的 STOMP 支持做到了这一点因此应用程序可以配置STOMP消息的最大尺寸而不考虑WebSocket服务器特定的消息尺寸。请记住如果有必要WebSocket消息的大小会被自动调整以确保它们至少可以承载16K的WebSocket消息。
下面的例子显示了一种可能的配置
Configuration
EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {Overridepublic void configureWebSocketTransport(WebSocketTransportRegistration registration) {registration.setMessageSizeLimit(128 * 1024);}// ...}下面的例子显示了前述例子的XML等效配置
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:websockethttp://www.springframework.org/schema/websocketxsi:schemaLocationhttp://www.springframework.org/schema/beanshttps://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/websockethttps://www.springframework.org/schema/websocket/spring-websocket.xsdwebsocket:message-brokerwebsocket:transport message-size131072 /!-- ... --/websocket:message-broker/beans 关于扩展的一个重要观点涉及到使用多个应用程序实例。目前你不能用 simple broker 做到这一点。但是当你使用全功能 broker如 RabbitMQ时每个应用程序实例都会连接到 broker并且从一个应用程序实例广播的消息可以通过 broker 广播给通过任何其他应用程序实例连接的 WebSocket 客户端。
二十二、 监控
当你使用 EnableWebSocketMessageBroker 或 websocket:message-broker 时关键的基础设施组件会自动收集统计信息和计数器从而提供对应用程序内部状态的重要洞察力。该配置还声明了一个 WebSocketMessageBrokerStats 类型的Bean它在一个地方收集所有可用的信息并默认每30分钟在 INFO 级别记录一次。这个Bean可以通过Spring的 MBeanExporter 导出到JMX以便在运行时查看例如通过JDK的 jconsole。下面的列表总结了可用的信息
Client WebSocket Sessions
Current
表示当前有多少个客户端会话该计数按WebSocket与HTTP流和轮询SockJS会话进一步细分。
Total
表示已经建立的总会话数量。
Abnormally Closed
Connect Failures
已建立的会话但在60秒内没有收到任何信息后被关闭。这通常是代理或网络问题的一个迹象。
Send Limit Exceeded
会话在超过配置的发送超时或发送缓冲区限制后关闭这可能发生在慢速客户端上见前一节。
Transport Errors
在发生传输错误后关闭的会话例如未能读取或写入WebSocket连接或HTTP请求或响应。
STOMP Frames
处理的CONNECT、CONNECTED和DISCONNECT帧的总数表示有多少客户端在STOMP层连接。请注意当会话被异常关闭或客户端关闭而没有发送DISCONNECT帧时DISCONNECT计数可能较低。
STOMP Broker Relay
TCP Connections
表示有多少代表客户WebSocket会话的TCP连接被建立到代理。这应该等于客户WebSocket会话的数量1个额外的共享 系统 连接用于从应用程序内发送消息。
STOMP Frames
代表客户端转发到 broker 处或从 broker 处接收的 CONNECT、CONNECTED 和 DISCONNECT 帧的总数。请注意无论客户WebSocket会话是如何关闭的DISCONNECT 帧都会被发送给 broker。因此较低的 DISCONNECT 帧计数表明 broker 正在主动关闭连接可能是因为心跳没有及时到达无效的输入帧或其他问题。
Client Inbound Channel
来自支持 clientInboundChannel 的线程池的统计数据提供了对传入消息处理的健康状况的洞察力。任务在这里排队是一个迹象表明应用程序可能太慢无法处理消息。如果有I/O绑定的任务例如缓慢的数据库查询对第三方REST API的HTTP请求等等考虑增加线程池大小。
Client Outbound Channel
来自支持 clientOutboundChannel 的线程池的统计数据提供了对向客户广播消息的健康状况的洞察力。任务在这里排队是一个迹象表明客户对消息的消费太慢了。解决这个问题的方法之一是增加线程池的大小以适应预期的并发慢速客户端的数量。另一个办法是减少发送超时和发送缓冲区大小的限制见上一节。
SockJS Task Scheduler
来自SockJS任务调度器的线程池的统计数据用于发送心跳。注意当心跳在STOMP级别上协商时SockJS的心跳被禁用。
二十三、 测试
当你使用Spring的基于 WebSocket 的 STOMP 支持时有两种主要方法来测试应用程序。第一种是编写服务器端测试以验证 controller 的功能和它们注解的消息处理方法。第二种是编写完整的端到端测试包括运行一个客户端和一个服务器。
这两种方法并不相互排斥。相反每种方法在整个测试策略中都有其位置。服务器端的测试更有针对性更容易编写和维护。另一方面端到端的集成测试更完整测试的内容更多但它们也更需要编写和维护。
服务器端测试的最简单形式是编写 controller 单元测试。然而这还不够有用因为 controller 所做的很多事情都取决于它的注解。纯粹的单元测试根本无法测试这些。
理想情况下被测试的 controller 应该在运行时被调用就像使用Spring MVC测试框架测试处理HTTP请求的 controller 的方法一样—也就是说不运行Servlet容器而是依靠Spring框架来调用注解的 controller。与Spring MVC测试一样你在这里有两种可能的选择要么使用 基于 context要么使用 独立 设置 在Spring TestContext框架的帮助下加载实际的Spring配置注入 clientInboundChannel 作为测试字段并使用它来发送 controller 方法所要处理的消息。 手动设置调用 controller 所需的最小Spring框架基础设施即 SimpAnnotationMethodMessageHandler并将 controller 的消息直接传递给它。
这两种设置情况在 股票投资组合 样本应用程序的测试中都有展示。
第二种方法是创建端到端的集成测试。为此你需要在嵌入式模式下运行WebSocket服务器并作为WebSocket客户端连接到它发送包含STOMP框架的WebSocket消息。 股票投资组合示例应用程序的测试 也展示了这种方法它使用Tomcat作为嵌入式WebSocket服务器和一个简单的STOMP客户端进行测试。
