java整合WebSocket

一、WebSocket介绍

1、简介

WebSocket协议通过在客户端和服务端之间提供全双工通信来进行Web和服务器的交互功能。在WebSocket应用程序中，服务器发布WebSocket端点，客户端使用url连接到服务器。建立连接后，服务器和客户端就可以互相发送消息。客户端通常连接到一台服务器，服务器接受多个客户端的连接。

2、优势

HTPP协议是基于请求响应模式，并且无状态的。HTTP通信只能由客户端发起，HTTP 协议做不到服务器主动向客户端推送信息。
如果我们想要查询当前的排队情况，只能是页面轮询向服务器发出请求，服务器返回查询结果。轮询的效率低，非常浪费资源（因为必须不停连接，或者 HTTP 连接始终打开）

3、服务端注解

@ServerEndpoint(“/websocket/{uid}”)
申明这是一个websocket服务；
需要指定访问该服务的地址，在地址中可以指定参数，需要通过{}进行占位；

@OnOpen
用法：public void onOpen(Session session, @PathParam(“uid”) String uid) throws IOException{}
该方法将在建立连接后执行，会传入session对象，就是客户端与服务端建立的长连接通道，通过@PathParam获取url中声明的参数；

@OnClose
用法：public void onClose() {}
该方法是在连接关闭后执行；

@OnMessage
用法：public void onMessage(String message, Session session) throws IOException {}
该方法用于接收客户端发送的消息；
message：发来的消息数据；
session：会话对象（也是长连接通道）；
发送消息到客户端；
用法：session.getBasicRemote().sendText(“hello,websocket.”);
通过session进行消息发送；

二、springboot整合

1、引入依赖

		<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>

2、配置

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

3、业务代码

>>群聊

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Component
@Slf4j
@ServerEndpoint(value = "/api/pushMessageMulti/{roomId}/{userId}",encoders = { ServerEncoder.class })
public class WebSocketServerMulti {

    /**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/
    private static int onlineCount = 0;
    /**concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。*/
    private static Map<String, WebSocketServerMulti> userMap = new ConcurrentHashMap<>();
    //存放房间对象
    private static Map<String, Set<WebSocketServerMulti>> roomMap = new ConcurrentHashMap<>();
    /**与某个客户端的连接会话，需要通过它来给客户端发送数据*/
    private Session session;
    /**接收userId*/
    private String userId = "";
    //存出当前群聊在线的人数（使用原因：roomMap中无法获取到当前用户id）
    private static Map<String, List<String>> multiUser = new ConcurrentHashMap<>();




    /**
     * 连接建立成
     * 功调用的方法
     */
    @OnOpen
    public void onOpen(Session session,@PathParam("roomId") String roomId , @PathParam("userId") String userId) throws IOException, EncodeException {
        synchronized (session){
            try {
                this.session = session;
                this.userId=userId;
                userMap.put(userId,this);
                if (!roomMap.containsKey(roomId)) {
                    Set<WebSocketServerMulti> set = new HashSet<>();
                    set.add(userMap.get(userId));
                    roomMap.put(roomId,set);

                    List<String> dd = new LinkedList<>();
                    dd.add(userId);
                    multiUser.put(roomId,dd);
                } else {
                    if(multiUser.get(roomId).contains(userId)){

                        log.info("存在：房间号:"+roomId+"用户连接:"+userId+",当前在线人数为:" + multiUser.get(roomId).size());
                    }else{
                        multiUser.get(roomId).add(userId);

                        roomMap.get(roomId).add(this);
                    }
                }
                System.out.println(multiUser.get(roomId).size());
                log.info("房间号:"+roomId+"用户连接:"+userId+",当前在线人数为:" + multiUser.get(roomId).size());
                Map<String,Object> map = new HashMap<>();
                map.put("online_num",multiUser.get(roomId).size());//在线人数
                map.put("online_list",roomMap.get(roomId));//人员列表
                map.put("roomId",roomId);//群id
                map.put("message","用户"+***+"加入群聊");//消息
				//自定义发送消息
                sendMessageObject(map,roomId);
            }catch (Exception e){
                e.printStackTrace();
            }

        }
    }

    /**
     * 连接关闭@PathParam("userId") String userId
     * 调用的方法
     */
    @OnClose
    public void onClose( @PathParam("roomId") String roomId,@PathParam("userId") String userId) {
        try {
            if (roomMap.containsKey(roomId)) {
                Set<WebSocketServerMulti> set = roomMap.get(roomId);
                Iterator<WebSocketServerMulti> it = set.iterator();
                while (it.hasNext()) {
                    if (it.next().userId.equals(userId)) {
                        it.remove();
                    }
                }

                multiUser.get(roomId).remove(userId);
                log.info("房间号:"+roomId+"用户退出:"+userId+",当前在线人数为:" + multiUser.get(roomId).size());
                Map<String,Object> map = new HashMap<>();
                map.put("online_num",multiUser.get(roomId).size());//在线人数
                map.put("online_list",roomMap.get(roomId));//人员列表
                map.put("roomId",roomId);//群id
                map.put("message","用户"+***+"加入群聊");//消息
				//自定义发送消息
                sendMessageObject(map,roomId);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 收到客户端消
     * 息后调用的方法
     * @param message
     * 客户端发送过来的消息
     **/
    @OnMessage
    public void onMessage(String message) {
        //注意，要给session加上同步锁，否则会出现多个线程同时往同一个session写数据，导致报错的情况。
        synchronized (session){
            //可以群发消息
            if(StringUtils.isNotBlank(message)){
                try {
                    //解析发送的报文
                    JSONObject jsonObject = JSONObject.parseObject(message);
                    //追加发送人(防止串改)
                    jsonObject.put("fromUserId",this.userId);
                    int chatType=jsonObject.getInteger("chatType");
                    String myUserId=jsonObject.getString("myUserId");
                    String toRoomId=jsonObject.getString("toRoomId");
                    log.info("房间号:"+toRoomId+"用户消息:"+userId+",报文:"+message);
                    sendMessageTo(message,toRoomId);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 群聊
     * @param message 消息
     * @param roomId 房间号
     */
    public void sendMessageTo(String message , String roomId) throws IOException {
        if (roomMap.containsKey(roomId)) {
            for (WebSocketServerMulti item : roomMap.get(roomId)) {
                    item.session.getAsyncRemote().sendText(message);
            }
        }
    }


    /**
     * @param error
     */
    @OnError
    public SystemResult onError(Throwable error) {

        log.error("用户错误:"+this.userId+",原因:"+error.getMessage());

        ChatError chatError = new ChatError();
        chatError.setUserId(Integer.valueOf(this.userId));
        chatError.setDetails(error.getMessage());
        chatError.setAddTime(new Date());
        chatErrorMapper.insert(chatError);
        SystemResult systemResult = new SystemResult();
        return systemResult.error(error.getMessage());

    }

    /**
     * 实现服务
     * 器主动推送
     */
    public void sendMessage(String message) {
        try {
            this.session.getBasicRemote().sendText(message);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 实现服务传送Object类型
     * 器主动推送
     */
    public void sendMessageObject(Object message,String roomId) {
        try {
            if (roomMap.containsKey(roomId)) {
                for (WebSocketServerMulti item : roomMap.get(roomId)) {
                    item.session.getBasicRemote().sendObject(message);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 获得此时的
     * 在线人数
     * @return
     */
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    /**
     * 在线人
     * 数加1
     */
    public static synchronized void addOnlineCount() {
        WebSocketServerMulti.onlineCount++;
    }

    /**
     * 在线人
     * 数减1
     */
    public static synchronized void subOnlineCount() {
        WebSocketServerMulti.onlineCount--;
    }

}

>>单人聊天

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;


@Component
@Slf4j
@ServerEndpoint("/api/pushMessageSolo/{userId}")
public class WebSocketServerSolo {

    /**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/
    private static int onlineCount = 0;
    /**concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。*/
    private static ConcurrentHashMap<String, WebSocketServerSolo> webSocketMap = new ConcurrentHashMap<>();
    /**与某个客户端的连接会话，需要通过它来给客户端发送数据*/
    private Session session;
    /**接收userId*/
    private String userId = "";

    /**
     * 连接建立成
     * 功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        this.userId=userId;
        if(webSocketMap.containsKey(userId)){
            webSocketMap.remove(userId);
            //加入set中
            webSocketMap.put(userId,this);
        }else{
            //加入set中
            webSocketMap.put(userId,this);
            //在线数加1
            addOnlineCount();
        }
        log.info("用户连接:"+userId+",当前在线人数为:" + getOnlineCount());
        sendMessage("连接成功");
    }

    /**
     * 连接关闭
     * 调用的方法
     */
    @OnClose
    public void onClose() {
        if(webSocketMap.containsKey(userId)){
            webSocketMap.remove(userId);
            //从set中删除
            subOnlineCount();
        }
        log.info("用户退出:"+userId+",当前在线人数为:" + getOnlineCount());
    }

    /**
     * 收到客户端消
     * 息后调用的方法
     * @param message
     * 客户端发送过来的消息
     **/
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("用户消息:"+userId+",报文:"+message);
        //可以群发消息
        //消息保存到数据库、redis
        if(StringUtils.isNotBlank(message)){
            try {
                //解析发送的报文
                JSONObject jsonObject = JSONObject.parseObject(message);
                //追加发送人(防止串改)
                jsonObject.put("fromUserId",this.userId);
                String toUserId=jsonObject.getString("toUserId");
                String myUserId=jsonObject.getString("myUserId");
                //传送给对应toUserId用户的websocket
                if(StringUtils.isNotBlank(toUserId)&&webSocketMap.containsKey(toUserId)){
                    webSocketMap.get(toUserId).sendMessage(message);
                }else{
                    //否则不在这个服务器上，发送到mysql或者redis
                    log.error("请求的userId:"+toUserId+"不在该服务器上");
                }
                if(StringUtils.isNotBlank(myUserId)&&webSocketMap.containsKey(myUserId)){
                    webSocketMap.get(myUserId).sendMessage(message);
                }else{
                    //否则不在这个服务器上，发送到mysql或者redis
                    log.error("请求的userId:"+myUserId+"不在该服务器上");
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }


    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {

        log.error("用户错误:"+this.userId+",原因:"+error.getMessage());
        error.printStackTrace();
    }

    /**
     * 实现服务
     * 器主动推送
     */
    public void sendMessage(String message) {
        try {
            this.session.getBasicRemote().sendText(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     *发送自定
     *义消息
     **/
    public static void sendInfo(String message, String userId) {
        log.info("发送消息到:"+userId+"，报文:"+message);
        if(StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)){
            webSocketMap.get(userId).sendMessage(message);
        }else{
            log.error("用户"+userId+",不在线！");
        }
    }

    /**
     * 获得此时的
     * 在线人数
     * @return
     */
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    /**
     * 在线人
     * 数加1
     */
    public static synchronized void addOnlineCount() {
        WebSocketServerSolo.onlineCount++;
    }

    /**
     * 在线人
     * 数减1
     */
    public static synchronized void subOnlineCount() {
        WebSocketServerSolo.onlineCount--;
    }

}

三、部署websocket项目问题

1、webSocket功能失效

本地开发的时候都可以正常使用，但是在部署到nginx代理服务器的时候发现报了错误，连不上，报错：Error in connection establishment: net::ERR_NAME_NOT_RESOLVED
发现是nginx服务器默认是不打开webSocket的功能的，这需要我们在nginx服务器上配置：

 location /test/ {
                proxy_pass http://test.com;
                proxy_redirect default;
                proxy_set_header Upgrade $http_upgrade; # allow websockets
                proxy_set_header Connection "upgrade";
                proxy_http_version 1.1;
                }

2、断线重连

如果nginx没有设置读取超时时间，websocket会一直断线重连，大约一分钟重连一次
可以设置长时间得超时时间，避免一直断线重连，避免消耗内存文章来源地址https://www.toymoban.com/news/detail-720058.html

location /test{
	    proxy_pass  http://test.com;
	    proxy_set_header Upgrade $http_upgrade; # allow websockets
    	proxy_set_header Connection "upgrade";
    	proxy_http_version 1.1;
        proxy_connect_timeout 60s;#l连接超时时间，不能设置太长会浪费连接资源
	    proxy_read_timeout 500s;#读超时时间
	    proxy_send_timeout 500s;#写超时时间
            index  index.html index.htm;
        }

到了这里，关于java整合WebSocket的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！