spring websocket stomp 集群模式

2021-05-26 本文已影响0人骑驴追地瓜

Websocket

一.websocket的使用场景
- websocket是一个双工协议，连接间可以进行实时动态交互，其实http+ajax的长短轮询也可以完成大部分场景下的动态交互问题，但是websocket有以下几个优势
  - 实时
  - 高频
  - 高容量
- 总结一下当需要实时，高频率且高容量交互的时候websocket是最佳的实现方式，比如:金融，游戏，聊天等。其它情况http+长或短轮询会是更好的实现方式，
二. Websocker协议
- TCP连接
  - 三次握手建立一个可信通道
  - 占用两个计算机之间的通信线路
  - 一台服务器能容纳的最多tcp连接数 4g内存可支持100w空tcp连接当有消息发送的时候会占用更多的内存当处理逻辑复杂，数据量大的时候会相应的占据更多cpu和内存
- 全双工
- 可以由一个HTTP发起在HTTP Header标识 Upgrade: websocket
  - 如：
```
GET /spring-websocket-portfolio/portfolio HTTP/1.1
Host: localhost:8080
Upgrade: websocket ①
Connection: Upgrade ② 
Sec-WebSocket-Key: Uc9l9TMkWGbHFD2qnFHltg==
Sec-WebSocket-Protocol: v10.stomp, v11.stomp
Sec-WebSocket-Version: 13
Origin: http://localhost:8080
```
- 返回的状态码，不再是200，而是101
  - 如:
```
  HTTP/1.1 101 Switching Protocols 
  Upgrade: websocket
  Connection: Upgrade
  Sec-WebSocket-Accept: 1qVdfYHU9hPOl4JYYNXF623Gzn0=
  Sec-WebSocket-Protocol: v10.stomp
```
- websocket虽然可以用http发起，但是本身属于低级协议，消息内容不包含任何语义，但是可以通过握手时的Sec-WebSocket-Protocol将协议升级到stomp
  - stomp与websocket关系
    - stomp Simple (or Streaming) Text Orientated Messaging Protocol 实际是一个简单流文本定向消息协议
    - stomp可以建立在websocket之上也可以建立在其它协议层上
      - stomp动作
```
  CONNECT、SEND、SUBSCRIBE、UNSUBSCRIBE、BEGIN、COMMIT、ABORT、ACK、NACK、DISCONNECT
```
      - 发送消息
```
  SEND
  destination:/queue/trade
  content-type:application/json
  content-length:44
  {“action”:”BUY”,”ticker”:”MMM”,”shares”,44}
```
      - 订阅消息
```
  SUBSCRIBE
  destination:/topic/price.stock.*
```
      - 服务器进行广播消息
```
  MESSAGE
  message-id:nxahklf6-1
  subscription:sub-1
  destination:/topic/price.stock.MMM
  {“ticker”:”MMM”,”price”:129.45}
```

三. stomp概念及配置

通过spring管理的stomp可以将消息发送给spring的controller或者是socket的broker
- broker spring作为websocket(stomp)服务端的代理所有socket请求都经过代理代理支持simp单机版和集群版（AMQP协议，RabbitMQ, ActiveMQ）
- controller spring的controller socket请求可以指向一个controller
架构图
- stomp-spring.jpg
配置
- img.png
- a.开启websocket自动配置
- b.配置stomp根地址，客户端通过根地址与服务端建立通信,如:ws://127.0.0.1:8080/stomp
- c.设置controller发送前缀/app
  - 如:
```
假设通道为/test.user
客户端发送
SEND 
destination:/app/test.user
{“action”:”BUY”,”ticker”:”MMM”,”shares”,44}
```
- d.设置代理通道前缀/topic 假设通道为/test.receive，客户端订阅时，如:SUBSCRIBE /topic/test.receive
  - 如:
```
假设通道为/test.receive
客户端订阅
SUBSCRIBE 
destination:/topic/test.receive
```
功能
- controller MessageMapping
  - 服务端可以配置controller,来接收并处理客户端发来的消息,method支持一些内置参数
  - controller.jpg
  - method.jpg
- 绑定user信息及鉴权
  - spring-stomp支持配置拦截器在连接订阅时对客户端做鉴权并根据鉴权信息绑定用户信息
  - auth-bind-user.jpg
- Sending Messages
  - 引入SimpMessagingTemplate，调用其send方法即可发送消息到客户端，点对点发送需要绑定user我们在，如下图
  - messageTemplate.jpg
  - user的监听通道默认为/user/topic/* 即当客户端user1订阅/user/topic/a的时候 SimpMessagingTemplate.convertAndSendToUser(user1,/topic/a)才能被user1收到，且必须经过上一步的user绑定
- @SendTo @SendToUser
  - 标注在controller方法上会将返回值自动发给订阅的通道
  - sendToUser.jpg
- MessageExceptionHandler
  - 消息异常处理可以通过@SendTo将异常发送到一个通道
  - exception.jpg

根据rabbitmq集群

rabbitmq配置

 sudo docker run  -p 35672:5672 -p 45672:15672 -p 51513:61613 -p 51514:61614 -d --hostname rabbitmq --name rabbitmq  -v /opt/docker/rabbitmq:/var/lib/rabbitmq/mnesia/rabbit@rabbitmq  -e RABBITMQ_DEFAULT_USER=rabbitmq -e RABBITMQ_DEFAULT_PASS=test  rabbitmq:3.8-rc-management
 sudo docker exec -it rabbitmq /bin/bash
 //开启stomp插件
 rabbitmq-plugins enable --offline rabbitmq_shovel rabbitmq_shovel_management rabbitmq_web_stomp

spring配置

@Configuration
@EnableWebSocketMessageBroker
public class StompBrokerConfig implements WebSocketMessageBrokerConfigurer {

  private static final String[] ENDPOINTS = new String[]{"/stomp"};

  /**
   * 发送频道   与订阅频道对应
   */
  private static final String[] APP_DES_PREFIXES = new String[]{"/app"};

  /**
   * 订阅频道
   */
  private static final String[] DES_PREFIXES = new String[]{"/topic"};
  /**
   * 当没有用户和服务可以处理消息的时候 统一向该地址发送广播
   */
  private static final String USER_DEST_BROADCAST = "/topic/unresolved-user";

  /**
   * 集群场景下的用户注册地址  使多服务端间用户透明
   */
  private static final String USER_REGISTRY_BROADCAST = "/topic/user-registry";

  @Override
  public void registerStompEndpoints(StompEndpointRegistry registry) {
      registry.addEndpoint(ENDPOINTS).setAllowedOrigins("*").withSockJS();
  }

  @Override
  public void configureMessageBroker(MessageBrokerRegistry registry) {
      //因为消息队列不适配/所以订阅消息统一用.
      registry.setPathMatcher(new AntPathMatcher("."));
      //设置client destination前缀
      registry.setApplicationDestinationPrefixes(APP_DES_PREFIXES);
      //普通模式 设置destination前缀
     // registry.enableSimpleBroker(DES_PREFIXES);
      //registry.enableSimpleBroker(DES_PREFIXES);
      //直接使用中继代理模式  进行集群适配
      registry.enableStompBrokerRelay(DES_PREFIXES)
              .setUserDestinationBroadcast(USER_DEST_BROADCAST)
              .setUserRegistryBroadcast(USER_REGISTRY_BROADCAST)
              .setRelayHost("rabbit-address")
              .setRelayPort("rabbit-stomp-port  51513")
              .setSystemLogin("rabbit-username")
              .setSystemPasscode("rabbit-password")
              .setClientLogin("rabbit-username")
              .setClientPasscode("rabbit-password")
              .setVirtualHost("vhost");
  }

  @Override
  public void configureClientInboundChannel(ChannelRegistration registration) {
      registration.interceptors(new UserHeaderInterceptor());
  }
}

spring websocket stomp 集群模式

Websocket

猜你喜欢

热点阅读