websocket实现QQ聊天一之一(服务端编写)

小灰灰 2021-06-11 15:10 509阅读 0赞

**下面介绍一个完整的websocket案例,案例实现socket获取HttpSession，和数据库用户实现绑定**

**一，后端涉及四个java类**

**![Center][]**

**1，websocket消息操作类WebSocketChat.java,先看一下java文件上面的注解**

@ServerEndpoint(value ="/webSocketChat.ws",encoders =\{ServerEncoder.class\},configurator=GetHttpSessionConfigurator.class)

**1.1注解中用到了三个属性**

**value**

不必多说

**encoders**

编码器：我在发送消息的时候，使用sendObject()方法，目的是发送json字符串到前端，解析相关的用户信息，因此需要使用encoders 属性，将数据编码为json。因此引出第二个相关类ServerEncoder.java，**ServerEncoder.java****将消息编码为json返回**

**configurator**

属性：她返回一个ServerEndpointConfig.Configurator对象，这个对象中有请求相关的所有信息，其中就有我们需要的HttpSession。因此引出第是三个第四个相关类GetHttpSessionConfigurator.java，RequestListener.java。**GetHttpSessionConfigurator.java:****说简单一点就是获取键值对的HttpSession，返回到ServerEndpointConfig.Configurator。****RequestListener.java****监听所有请求，将所有request请求都携带上httpSession，如果不实现监听，那么GetHttpSessionConfigurator获取的HttpSession，就是null的，会报错：java.lang.NullPointerException，这个错误信息，大家最熟悉不过了。**

**说明：注解非常简单可以直接使用注解@WebListener，也可以再web.xml配置监听，这里不做演示**

好了，介绍到这里，直接上代码。

**WebSocketChat.java**

package com.zlxls.information;
    
    import com.common.Constants;
    import com.common.model.Admin;
    import com.common.model.SocketMsg;
    import com.jfinal.plugin.activerecord.Db;
    import com.zlxls.util.Validate;
    import java.io.IOException;  
    import java.util.Date;
    import java.util.HashMap;
    import java.util.Map;
    import javax.servlet.http.HttpSession;
    import javax.websocket.EncodeException;
    import javax.websocket.EndpointConfig;
      
    import javax.websocket.OnClose;
    import javax.websocket.OnError;
    import javax.websocket.OnMessage;
    import javax.websocket.OnOpen;
    import javax.websocket.Session;
    import javax.websocket.server.ServerEndpoint;
    /**
     * TODO(在这里用一句话描述这个类的作用) 
     * @ClassNmae：ChatEntpoint   
     * @author zlx-雄雄
     * @date    2017-8-30 8:28:53
     * 
     */
    @ServerEndpoint(value ="/webSocketChat.ws",encoders ={ServerEncoder.class},configurator=GetHttpSessionConfigurator.class)
    public class WebSocketChat{
        /** 
         * 一个提供原子操作的Integer的类。在Java语言中，++i和i++操作并不是线程安全的， 
         * 在使用的时候，不可避免的会用到synchronized关键字。 而AtomicInteger则通过一种线程安全的加减操作接口。 
         */  
        private static final Map<String, Session> connections = new HashMap<>();
        Admin admin;
        /** 
         * 传输信息过程中调用方法 
         * @param message 
         * @param session 
         * @throws java.io.IOException 
         */  
        @OnMessage
        public void onMessage(String message, Session session) throws IOException {
            String[] messages = message.split("#");  
            String msg = messages[0];
            SocketMsg socketMsg = new SocketMsg(); 
            socketMsg.set("from_user", admin.getAdminId()).set("from_user_name", admin.getAdminName()).set("time",new Date()).set("msg", msg);
            if(messages.length==2){
                //点对点发送 
                Admin to_admin = Admin.dao.findById(messages[1]);
                socketMsg.set("to_user", to_admin.getAdminId()).set("to_user_name", to_admin.getAdminName());
            }else{
                socketMsg.set("to_user","").set("to_user_name","");
            }
            if(socketMsg.save()){
                System.err.println("保存消息成功");
                broadcast(socketMsg);  
            }else{
                session.getBasicRemote().sendText("消息发送失败！");
            }
        }  
        /**
        * 发送点对点信息
        */
        private void broadcastOne(SocketMsg socketMsg){
            Session session = connections.get(socketMsg.getToUser());
             try {
                    if(session!=null && session.isOpen()){
                        connections.get(socketMsg.getToUser()).getBasicRemote().sendObject(socketMsg);//发给好友
                        Db.update("update socket_msg set status=0 where id=?",socketMsg.getId());//如果连接未关闭就更改为已读状态
                    }
                    connections.get(socketMsg.getFromUser()).getBasicRemote().sendObject(socketMsg);//发给自己
                } catch (IOException | EncodeException e) {
                    connections.remove(socketMsg.getToUser());
                    e.printStackTrace();
                    try {
                        connections.get(socketMsg.getToUser()).close();
                    } catch (IOException e1) {
                        e1.printStackTrace();
                    }
                }
        }
        /**
        * 发送广播
        */
        private void broadcastAll(SocketMsg socketMsg){
            connections.entrySet().stream().forEach((entry) -> {
                try {
                    synchronized (entry) {
                        entry.getValue().getBasicRemote().sendObject(socketMsg);
                    }
                } catch (IOException | EncodeException e) {
                    connections.remove(entry.getKey());
                    e.printStackTrace();
                    try {
                        entry.getValue().close();
                    } catch (IOException e1) {
                        e1.printStackTrace();
                    }
                }
            });
        }
        /** 
         * 消息广播 
         * 通过connections，对所有其他用户推送信息的方法 
         * @param msg
         */   
        private void broadcast(SocketMsg socketMsg){  
            if(Validate.isNotNull(socketMsg.getToUser())){
                broadcastOne(socketMsg);
            }else{
                broadcastAll(socketMsg);
            }
        }  
        /** 
         * 连接成功后，写入队列
         * @param msg
         */   
        private void setMap(Session session,EndpointConfig config){  
            
            HttpSession httpSession= (HttpSession)config.getUserProperties().get(HttpSession.class.getName());
            
            admin = (Admin)httpSession.getAttribute(Constants.SESSION_ADMIN);
            
            connections.put(admin.getAdminId(),session);
        }  
        /** 
         * 创建连接时间调用的方法 
         * @param session
         * @param config
         */  
        @OnOpen
        public void onOpen(Session session,EndpointConfig config) {
            System.out.println("服务已连接");
            setMap(session, config);
            
        }  
        /** 
         * 链接关闭时调用方法 
         */  
        @OnClose
        public void onClose() {  
            System.out.println("服务已关闭");
        }  
        /** 
         * 发生错误是调用方法 
         * @param t 
         * @throws Throwable 
         */  
        @OnError  
        public void onError(Throwable t) throws Throwable {  
            System.out.println("错误: " + t.toString());  
        }  
    }

ServerEncoder.java

package com.zlxls.information;
    
    import com.alibaba.fastjson.JSON;
    import com.common.model.SocketMsg;
    import javax.websocket.EncodeException;
    import javax.websocket.Encoder;
    import javax.websocket.EndpointConfig;
    
    /**
     * 配置WebSocket解码器，用于发送请求的时候可以发送Object对象，实则是json数据
     * sendObject()
     * @ClassNmae：ServerEncoder   
     * @author zlx-雄雄
     * @date    2017-11-3 15:47:13
     * 
     */
    public class ServerEncoder implements Encoder.Text<SocketMsg> {  
      
        @Override  
        public void destroy() {  
            // TODO Auto-generated method stub  
      
        }  
      
        @Override  
        public void init(EndpointConfig arg0) {  
            // TODO Auto-generated method stub  
      
        }  
      
        @Override  
        public String encode(SocketMsg socketMsg) throws EncodeException {  
            try {  
                return JSON.toJSONString(socketMsg);  
            } catch (Exception e) {  
                // TODO Auto-generated catch block  
                e.printStackTrace();  
                return "";  
            }  
        }  
      
    }

GetHttpSessionConfigurator.java

package com.zlxls.information;
    
    import javax.servlet.http.HttpSession;
    import javax.websocket.HandshakeResponse;
    import javax.websocket.server.HandshakeRequest;
    import javax.websocket.server.ServerEndpointConfig;
    import javax.websocket.server.ServerEndpointConfig.Configurator;
    /**
     * 由于websocket的协议与Http协议是不同的，
     * 所以造成了无法直接拿到session。
     * 但是问题总是要解决的，不然这个websocket协议所用的场景也就没了
     * 重写modifyHandshake，HandshakeRequest request可以获取httpSession
     * @ClassNmae：GetHttpSessionConfigurator   
     * @author zlx-雄雄
     * @date    2017-11-3 15:47:13
     * 
     */
    public class GetHttpSessionConfigurator extends Configurator{
        @Override
        public void modifyHandshake(ServerEndpointConfig sec,HandshakeRequest request, HandshakeResponse response) {
            
            HttpSession httpSession=(HttpSession) request.getHttpSession();
            
            sec.getUserProperties().put(HttpSession.class.getName(),httpSession);
            
        }
    }

**RequestListener.java**

package com.zlxls.information;
    
    import javax.servlet.ServletRequestEvent;
    import javax.servlet.ServletRequestListener;
    import javax.servlet.annotation.WebListener;
    import javax.servlet.http.HttpServletRequest;
    
    /**
     * http://www.cnblogs.com/zhuxiaojie/p/6238826.html
     * 配置监听器，将所有request请求都携带上httpSession
     * 用于webSocket取Session
     * @ClassNmae：RequestListener   
     * @author zlx-雄雄
     * @date    2017-11-4 11:27:33
     * 
     */
    @WebListener
    public class RequestListener implements ServletRequestListener {
        @Override
        public void requestInitialized(ServletRequestEvent sre)  {
            
            //将所有request请求都携带上httpSession
            ((HttpServletRequest) sre.getServletRequest()).getSession();
            
        }
        public RequestListener() {}
    
        @Override
        public void requestDestroyed(ServletRequestEvent arg0)  {}
    }

当然，这里服务端也就完成了，但是需要大家在自己项目的控制器中添加一个登陆方法，用于写入HttpSession。

更多[2][] **分享到：**

[Center]: /images/20210516/bc4c14b718af41b4be60e4ddb7acfbef.png
[2]: http://bbs.csdn.net/topics/390241173#