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 java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.AbstractWebSocketHandler;
import org.springframework.web.socket.sockjs.transport.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) FieldUtils.readField(session, "webSocketSession", true)).getId();
} catch (IllegalAccessException e) {
throw new RuntimeException("get sessionId error");
}
}
return session.getId();
}
/**
* webSocket連接創建后調用
*/
@Override
public void afterConnectionEstablished(WebSocketSession session) {
// 獲取參數
String user = String.valueOf(session.getAttributes().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) {
sessionMap.remove(getSessionId(session));
}
/**
* 連接關閉會調用
*/
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
sessionMap.remove(getSessionId(session));
}
@Override
public boolean supportsPartialMessages() {
return false;
}
/**
* 后端發送消息
*/
public void sendMessage(String user, String message) {
String sessionId = userMap.get(user);
WebSocketSession session = sessionMap.get(sessionId);
try {
session.sendMessage(new TextMessage(message));
} catch (IOException e) {
e.printStackTrace();
}
}
}
自定義攔截器：
public class WebSocketInterceptor implements HandshakeInterceptor {
/**
* handler處理前調用,attributes屬性最終在WebSocketSession里,可能通過webSocketSession.getAttributes().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 = serverHttpRequest.getServletRequest().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");
ws.onopen = function() {
// Web Socket 已連接上，使用 send() 方法發送數據
ws.send("測試發送");
console.log('open');
};
ws.onmessage = function (e) {
console.log('message', e.data);
};
ws.onclose = function() {
console.log('close');
};
sockJs連接
<script src="//cdn.jsdelivr.net/sockjs/1.0.0/sockjs.min.js"></script>
var sock = new SockJS('http://localhost:8080/sock-js?user=Tony');
sock.onopen = function() {
sock.send("測試發送");
console.log('open');
};
sock.onmessage = function(e) {
console.log('message', e.data);
};
sock.onclose = function() {
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 = RequestContext.getCurrentContext();
HttpServletRequest request = context.getRequest();
String upgradeHeader = request.getHeader("Upgrade");
if (null == upgradeHeader) {
upgradeHeader = request.getHeader("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;
location / {
proxy_pass http://[服務器組名];
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;
}
}
前端連接示例：
sockJs連接
<script src="//cdn.jsdelivr.net/sockjs/1.0.0/sockjs.min.js"></script>
var sock = new SockJS('http://(網關路徑)/(服務名稱)/sock-js?user=Tony');
sock.onopen = function() {
sock.send("測試發送");
console.log('open');
};
sock.onmessage = function(e) {
console.log('message', e.data);
};
sock.onclose = function() {
console.log('close');
};
webSocket連接
// 打開一個 web socket
var ws = new WebSocket("ws://(網關路徑)/(服務名稱)/websocket?user=Tony");
ws.onopen = function() {
// Web Socket 已連接上，使用 send() 方法發送數據
ws.send("測試發送");
console.log('open');
};
ws.onmessage = function (e) {
console.log('message', e.data);
};
ws.onclose = function() {
console.log('close');
};
