1.添加websocket依赖
org.springframework.boot spring-boot-starter-websocket
2.springboot的启动类注册bean
WebSocket配置类 @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); }
3.websocket操作类:
import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import javax.websocket.*; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; @ServerEndpoint(value = "/websocket/{id}") @Component public class Websocket{ private static final Logger LOGGER = LoggerFactory.getLogger(Websocket.class); private static AtomicInteger onlineCount = new AtomicInteger(0); private static ConcurrentHashMapmesSessionMap = new ConcurrentHashMap<>(); private static ConcurrentHashMap mesWebsocketMap = new ConcurrentHashMap<>(); @OnOpen public void onOpen(@PathParam("id") String id, Session session) { onlineCount.incrementAndGet(); mesSessionMap.put(id, session); mesWebsocketMap.put(id, this); LOGGER.info("有新连接加入:{},当前在线人数为:{}", session.getId(), onlineCount.get()); } @OnClose public void onClose(@PathParam("id") String id, Session session) { onlineCount.decrementAndGet(); mesSessionMap.remove(id); mesWebsocketMap.remove(id); LOGGER.info("有一连接关闭:{},当前在线人数为:{}", session.getId(), onlineCount.get()); } @OnError public void onError(Session session, Throwable error) { LOGGER.error("发生错误", error); error.printStackTrace(); } public void sendMessage(String id, String json) { Session session = mesSessionMap.get(id); if (session != null) { session.getAsyncRemote().sendText(json); } } public static Websocket getWebSockets(String id) { Websocket webSockets = mesWebsocketMap.get(id); return webSockets; } @OnMessage public void onMessage(@PathParam("id") String id, String test, Session session) { System.out.println("来自客户端的消息:" + id + ";" + test); mesWebsocketMap.get(id).sendMessage(id, "json"); } }
4.nginx支撑websocket配置,其中ip和port为自身的微服务ip和端口。(若该服务中又权限控制,做路劲过滤。)
location /api/websocket/ { proxy_pass http://ip:port/; proxy_http_version 1.1; proxy_set_header Upgrade websocket; proxy_set_header Connection upgrade; proxy_connect_timeout 75; proxy_read_timeout 1800s; proxy_send_timeout 1800s; add_header 'Access-Control-Allow-Origin' '$http_origin' always; add_header 'Access-Control-Allow-Credentials' 'true' always; add_header 'Access-Control-Allow-Methods' 'GET,POST,OPTIONS,PUT,DELETE' always; add_header 'Access-Control-Allow-Headers' 'DNT,X-CustomHeader,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Content-Range,Range' always; add_header 'Access-Control-Expose-Headers' 'DNT,X-CustomHeader,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Content-Range,Range' always; if ($request_method = 'OPTIONS') { add_header 'Access-Control-Allow-Origin' '$http_origin' always; add_header 'Access-Control-Allow-Credentials' 'true' always; add_header 'Access-Control-Allow-Methods' 'GET,POST,OPTIONS,PUT,DELETE' always; add_header 'Access-Control-Allow-Headers' 'Pragma,DNT,X-CustomHeader,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Content-Range,Range' always; add_header 'Access-Control-Expose-Headers' 'DNT,X-CustomHeader,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Content-Range,Range' always; add_header 'Access-Control-Max-Age' '1728000' always; add_header 'Content-Type' 'text/plain; charset=utf-8' always; add_header 'Content-Length' 0 always; return 204; } }
5.1测试如下:测试工具为Apipost
建立连接如图:
5.2测试发送接口(对接websocket的onMessage接口)