springCloud后端使⽤webSocket向前端推送消息
1、webSocket
webSocket长连接是⼀种在单个tcp连接上进⾏全双⼯通信的协议，允许双向数据推送。⼀般微服务提供的restful API只是对前端请求做出相应。使⽤webSocket可以实现后端主动向前端推送消息。
2、springboot使⽤webSocket
1、pom⽂件添加依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
2、添加webSocket配置
@Configuration
@EnableWebSocket
public class WebSocketAutoConfig implements WebSocketConfigurer {
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
// webSocket通道
// 指定处理器和路径
registry.addHandler(new WebSocketHandler(), "/websocket")
// 指定⾃定义
.addInterceptors(new WebSocketInterceptor())
// 允许跨域
.setAllowedOrigins("*");
/
/ sockJs通道
registry.addHandler(new WebSocketHandler(), "/sock-js")
.addInterceptors(new WebSocketInterceptor())
.setAllowedOrigins("*")
// 开启sockJs⽀持
.withSockJS();
}
}
处理器代码：
import java.io.IOException;
import java.util.Map;
import urrent.ConcurrentHashMap;
import org.flect.FieldUtils;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.AbstractWebSocketHandler;
import org.springframework.web.ansport.session.WebSocketServerSockJsSession;
@Component
public class WebSocketHandler extends AbstractWebSocketHandler {
/**
* 存储sessionId和webSocketSession
* 需要注意的是，webSocketSession没有提供⽆参构造，不能进⾏序列化，也就不能通过redis存储
* 在分布式系统中，要想别的办法实现webSocketSession共享
*/
private static Map<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();
private static Map<String, String> userMap = new ConcurrentHashMap<>();
/**
* 获取sessionId
*/
private String getSessionId(WebSocketSession session) {
if (session instanceof WebSocketServerSockJsSession) {
// sock js 连接
try {
return ((WebSocketSession) adField(session, "webSocketSession", true)).getId();
} catch (IllegalAccessException e) {
throw new RuntimeException("get sessionId error");
}
}
Id();
}
/**
* webSocket连接创建后调⽤
*/
@Override
public void afterConnectionEstablished(WebSocketSession session) {
// 获取参数
String user = String.Attributes().get("user"));
String sessionId = getSessionId(session);
userMap.put(user, getSessionId(session));
sessionMap.put(sessionId, session);
}
/**
* 接收到消息会调⽤
*/
@Override
public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
if (message instanceof TextMessage) {
} else if (message instanceof BinaryMessage) {
} else if (message instanceof PongMessage) {
} else {
System.out.println("Unexpected WebSocket message type: " + message);
}
}
/**
* 连接出错会调⽤
*/
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) {
}
/**
* 连接关闭会调⽤
*/
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
}
@Override
public boolean supportsPartialMessages() {
return false;
}
/**
* 后端发送消息
*/
public void sendMessage(String user, String message) {
String sessionId = (user);
WebSocketSession session = (sessionId);
try {
session.sendMessage(new TextMessage(message));
} catch (IOException e) {
e.printStackTrace();
}
}
}
⾃定义：
public class WebSocketInterceptor implements HandshakeInterceptor {
/**
* handler处理前调⽤,attributes属性最终在WebSocketSession⾥，可能通过Attributes().get(key值)获得
*/
@Override
public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) { if (request instanceof ServletServerHttpRequest) {
ServletServerHttpRequest serverHttpRequest = (ServletServerHttpRequest) request;
// 获取请求路径携带的参数
String user = ServletRequest().getParameter("user");
attributes.put("user", user);
return true;
} else {
return false;
}
}
@Override
public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
}
}
2、前端连接webSocket
WebSocket对象是H5新增的对象，低版本浏览器可能不⽀持，可以使⽤sockJs的⽅式连接
webSocket连接
// 打开⼀个 web socket
var ws = new WebSocket("ws://localhost:8080/websocket?user=Tony");
// Web Socket 已连接上，使⽤ send() ⽅法发送数据
ws.send("测试发送");
console.log('open');
};
console.log('message', e.data);
};
console.log('close');
};
sockJs连接
<script src="//cdn.jsdelivr/sockjs/1.0.0/sockjs.min.js"></script>
var sock = new SockJS('localhost:8080/sock-js?user=Tony');
sock.send("测试发送");
console.log('open');
};
console.log('message', e.data);
};
console.log('close');
3、springCloud使⽤zuul转发webSocket连接
zuul⽹关还需要做额外的配置，添加⾃定义过滤器：
@Component
public class WebSocketFilter extends ZuulFilter {
@Override
public String filterType() {
return "pre";
}
@Override
public int filterOrder() {
return 0;
}
@Override
public boolean shouldFilter() {
return true;
}
@Override
public Object run() {
RequestContext context = CurrentContext();
HttpServletRequest request = Request();
String upgradeHeader = Header("Upgrade");
if (null == upgradeHeader) {
upgradeHeader = Header("upgrade");
}
if (null != upgradeHeader && "websocket".equalsIgnoreCase(upgradeHeader)) {
context.addZuulRequestHeader("connection", "Upgrade");
}
return null;
}
}
配置⽂件添加配置：
hystrix:
command:
default:
execution:
isolation:
thread:
timeoutInMilliseconds: 60000 #设置API⽹关中路由转发请求的HystrixCommand执⾏超时时间ribbon:
ConnectTimeout: 3000 #设置路由转发请求的时候，创建请求连接的超时时间
ReadTimeout: 60000 #⽤来设置路由转发请求的超时时间
4、⽹关使⽤springCloud Gateway⽆需做任何处理
5、若使⽤了Nginx，需要添加配置
配置⽰例：
map $http_upgrade $connection_upgrade {
default upgrade;
'' close;
}
# ⽹关只有⼀个，写⼀个⽹关地址就⾏了
upstream [服务器组名] {
server [IP地址]:[端⼝号];
server [IP地址]:[端⼝号];
...
}
server {
listen 80;
前端websocket怎么用location / {
proxy_pass [服务器组名];
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;
}
}
前端连接⽰例：
sockJs连接
<script src="//cdn.jsdelivr/sockjs/1.0.0/sockjs.min.js"></script>
var sock = new SockJS('(⽹关路径)/(服务名称)/sock-js?user=Tony');
sock.send("测试发送");
console.log('open');
};
console.log('message', e.data);
};
console.log('close');
};
webSocket连接
/
/ 打开⼀个 web socket
var ws = new WebSocket("ws://(⽹关路径)/(服务名称)/websocket?user=Tony");
// Web Socket 已连接上，使⽤ send() ⽅法发送数据
ws.send("测试发送");
console.log('open');
};
console.log('message', e.data);
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。
发表评论