Web在线聊天室(8) --- 服务器中换收发消息

发布于2021-05-30 12:12 阅读(1107) 评论(0) 点赞(26) 收藏(3)

搭建服务器框架

客户端

        sendMessage() {
          let message = {
            userId: this.user.userId,
            nickName: this.user.nickName,
            channelId: this.curChannelId,
            content: this.curInputMessage,
          }
          this.websocket.send(JSON.stringify(message));
          console.log("send success! " + message.content);
          this.curInputMessage = "";
        },
        initWebSocket() {
          // 登陆成功后, 初始化 WebSocket 对象
          if('WebSocket' in window){
            this.websocket = new WebSocket("ws://localhost:8080/java_chatroom/message/" + this.user.userId);
            console.log("link success")
          }else{
              alert('Not support websocket')
          }
           //连接发生错误的回调方法
          this.websocket.onerror = function () {
            alert("连接发生错误!");
          };

          //连接成功建立的回调方法
          this.websocket.onopen = function (event) {
            console.log("连接建立成功");
          }

          //接收到消息的回调方法
          this.websocket.onmessage = function (event) {
            console.log("recv success! " + event.data);
            let message = JSON.parse(event.data);
            // let length = app.messages.length;
            // app.messages[length] = message;
            app.messages.push(message);
            // 不要忘了需要筛选指定的消息
            app.curChannelMessages = app.messages.filter((message, i) => {
              if (message.channelId == app.curChannelId) {
                return true;
              }
              return false;
            });
          }

          //连接关闭的回调方法
          this.websocket.onclose = function () {
            console.log("服务器断开连接");
          }

          //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
          window.onbeforeunload = function () {
            console.log("客户端断开连接");
            websocket.close();
          }
        },

服务器端

这里用到WebSocket技术，先写一个服务器端的框架，然后再进行补充填写

package org.example.servlet;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

/**
 * Created with IntelliJ IDEA.
 * Description:服务器websocket
 * User: starry
 * Date: 2021 -05 -27
 * Time: 8:54
 */

@ServerEndpoint("/message/{userId}")
public class MessageWebsocket {

    @OnOpen
    public void onOpen(@PathParam("userId") Integer userId, Session session) {
        System.out.println("建立连接" + userId);
    }

    @OnMessage
    public void onMessage(Session session, String message) {
        System.out.println("接收到的消息：" + message);
    }

    @OnClose
    public void onClose(Session session) {
        System.out.println("关闭连接");
    }

    @OnError
    public void onError(Throwable throwable) {
        System.out.println("出错了");
        throwable.printStackTrace();
    }

}

实现效果

我们在页面上输入hello，然后点击发送
在这里插入图片描述
查看本地idea上输出的信息

没有问题，建立连接后，并收到了消息的所有属性，内容为“hello”

收发消息

编写messageCenter保存websocket需要的信息

编写一个model类messageCenter，用来把所有的在线用户的session保存在map集合中

客户端登录后与服务器端建立连接，这里是websocket的session进行建立
客户端发送消息的时候，把要发送的message对象发送给服务器端
服务器端调用onmessage方法通过websocket的session转发此条消息给所有在线的客户端
前端接收到消息的回调函数
前端把返回来的消息根据频道id来过滤数据并显示到对应频道房间内

package org.example.model;

import javax.websocket.Session;
import java.io.IOException;
import java.util.Enumeration;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description:保存websocket需要的信息（所有客户端session）
 * User: starry
 * Date: 2021 -05 -27
 * Time: 9:48
 */
public class MessageCenter {

    /**
     * 支持线程安全的map结构，并且满足高并发（读写，读读并发，写写互斥，加锁粒度）
     */
    private static final ConcurrentHashMap<Integer, Session>  clients = new ConcurrentHashMap<>();

    /**
     * websocket建立连接时，添加用户id和客户端session，保存起来
     */
    public static void addOnlineUser(Integer userId, Session session) {
        clients.put(userId,session);
    }

    /**
     * 关闭websocket连接，和出错时，删除客户端session
     */
    public static void delOnlineUser(Integer userId) {
        clients.remove(userId);
    }

    /**
     * 接收到某用户的消息时，转发到所有客户端
     */
    public static void sendMessage(String message) {
        try {
            //遍历map的所有数据，然后转发到所有session对象中
            Enumeration<Session> e = clients.elements();
            while (e.hasMoreElements()) {
                Session session = e.nextElement();
                session.getBasicRemote().sendText(message);
            }
        } catch (IOException ioException) {
            ioException.printStackTrace();
        }
    }

}

在这里插入图片描述
但是存在一个消息转发所有客户端，存在新能问题如果接收到的信息数量m很多，同时在线的用户数量n也很多，那么要转发的次数就是m*n次，每个接收消息都是一个线程，都要等待websocket中的onmessage回调方法执行完，性能差

优化（使用阻塞队列的方式解决：并行并发的执行任务提交和执行任务）

服务器端调用该类

package org.example.servlet;

import org.example.model.MessageCenter;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

/**
 * Created with IntelliJ IDEA.
 * Description:服务器websocket
 * User: starry
 * Date: 2021 -05 -27
 * Time: 8:54
 */

@ServerEndpoint("/message/{userId}")
public class MessageWebsocket {

    @OnOpen
    public void onOpen(@PathParam("userId") Integer userId, Session session) {
        // 1.把每个客户端的session都保存起来，之后转发消息到所有客户端要用
        MessageCenter.addOnlineUser(userId,session);
        // 2.查询本客户端（用户）上次登录前的消息（数据库查）
        System.out.println("建立连接" + userId);
    }

    @OnMessage
    public void onMessage(Session session, String message) {
        // 1.遍历保存的所有session，每个都发送消息
        // 2.消息还要保存在数据库
        MessageCenter.sendMessage(message);
        System.out.println("接收到的消息：" + message);
    }

    @OnClose
    public void onClose(Session session) {
        // 本客户端关闭连接，要在之前保存的session集合中，删除
        System.out.println("关闭连接");
    }

    @OnError
    public void onError(Throwable throwable) {
        System.out.println("出错了");
        throwable.printStackTrace();
        //和关闭连接的操作一样
    }

}