socket心跳重连机制

港控/mmm° 2021-11-17 09:18 472阅读 0赞

先看没有心跳重连 的：[https://mp.csdn.net/postedit/87882743][https_mp.csdn.net_postedit_87882743]

package im.qingtui.meeting.socket;
    
    import im.qingtui.meeting.constants.ErrorCodeConstants;
    import im.qingtui.meeting.constants.exception.SocketException;
    import im.qingtui.meeting.dao.UserInfoMapper;
    import im.qingtui.meeting.model.Token;
    import im.qingtui.meeting.model.dto.emun.ClientType;
    import im.qingtui.meeting.model.msg.SocketSession;
    import im.qingtui.meeting.service.CommonService;
    import im.qingtui.meeting.utils.SpringContextUtils;
    import im.qingtui.meeting.utils.StringUtil;
    import im.qingtui.meeting.utils.StringUtils;
    import im.qingtui.platform.common.SpringFactory;
    import java.io.IOException;
    import java.util.List;
    import java.util.Set;
    import java.util.concurrent.CopyOnWriteArraySet;
    import javax.annotation.PostConstruct;
    import javax.servlet.annotation.WebServlet;
    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.PathParam;
    import javax.websocket.server.ServerEndpoint;
    import lombok.extern.slf4j.Slf4j;
    import net.sf.json.JSONObject;
    import org.apache.commons.lang3.EnumUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.aspectj.EnableSpringConfigured;
    import org.springframework.stereotype.Component;
    
    /**
     * socket全双工服务
     *
     * @author qiaofeng
     */
    @Slf4j
    @ServerEndpoint(value = "/socket/meeting/{openId}/{client}/{accessToken}")
    @Component
    public class SocketService {
    
        @Autowired
        private UserInfoMapper userInfoMapper = (UserInfoMapper) SpringContextUtils.getSpringBean("userInfoMapper");
    
        //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
        private static int onLineCount = 0;
    
        //连接的所有session
        private static Set<SocketSession> connectionSet = new CopyOnWriteArraySet<>();
    
        //初始化标志
        private static boolean isHeartStart = false;
    
        @OnOpen
        public synchronized void onOpen(@PathParam("openId") String openId, @PathParam("client") String client, @PathParam("accessToken") String token,
            Session session, EndpointConfig config) {
            checkSocketParam(openId, client);
            checkToken(token);
            SocketSession socketSession = getSocketInfo(openId, client);
            if (socketSession == null) {
                connectionSet.add(new SocketSession(openId, ClientType.getClientType(client), session));
            } else {
                socketSession.setHeart(true);
                socketSession.setSession(session);
                log.info("用户openId{}在客户端为client{}再次连接，现在连接数为{}", openId, client, getOnlineCount());
            }
            if (!isHeartStart && connectionSet.size() == 1) {
                isHeartStart = true;
                startHeart();
            }
        }
    
        @OnMessage
        public synchronized void onMessage(@PathParam("openId") String openId, @PathParam("client") String client, String message, Session session) {
            checkSocketParam(openId, client);
            JSONObject jsonObject = JSONObject.fromObject(message);
            if (jsonObject.has("secret") && "ping".equals(jsonObject.getString("secret")) && jsonObject.has("type") && "qingtui".equals(jsonObject.getString("type"))) {//心跳
                log.info("收到{}的心跳message{}", openId, message);
                //如果收到了心跳 这里设置isHeart为true
                SocketSession socketEntity = getSocketInfo(openId, client);
                if (null != socketEntity) {
                    socketEntity.setTimeStr(System.currentTimeMillis());
                    socketEntity.setHeart(true);
                }
            }
            //普通对话
        }
    
        /**
         * 给部分人发送消息
         *
         * @param openIdList 用户openId集合
         * @param msg 消息体信息 { "cmd": 1, “msg”:{ “meetingId”：””, “openId”: ””,   //改变状态人的openId “attendStatus”:””,  //参加状态（0待，1确认参加、2不参加     即请假 “leaveReason”：“请假理由”
         * } }
         */
        public void sendMsgToUser(List<String> openIdList, String msg) {
            if (connectionSet.size() == 0 || StringUtil.isEmptyList(openIdList)) {
                return;
            }
            for (String toOpenId : openIdList) {
                for (SocketSession socketSession : connectionSet) {
                    if (socketSession.getOpenId().equals(toOpenId)) {
                        try {
                            socketSession.getSession().getBasicRemote().sendText(msg);
                        } catch (Exception e) {
                            log.error("socket推送消息失败" + e);
                            throw new SocketException(ErrorCodeConstants.SOCKET_ERR);
                        }
                    }
                }
            }
    
        }
    
    
        @OnError
        public synchronized void onError(@PathParam("openId") String openId, @PathParam("client") String client, Session session, Throwable error)
            throws IOException {
            checkSocketParam(openId, client);
            onClose(openId, client);
        }
    
        @OnClose
        public synchronized void onClose(@PathParam("openId") String openId, @PathParam("client") String client) {
            log.info("openId为{}，客户端类型为{}退出了链接", openId, client);
            if (!ClientType.checkParamIsHave(client) || StringUtil.isEmpty(openId) || userInfoMapper.getUserInfoByOpenId(openId) == null) {
                throw new SocketException(ErrorCodeConstants.PARAM_ERROR);
            }
            SocketSession socketSession = getSocketInfo(openId, client);
            if (socketSession != null) {
                connectionSet.remove(socketSession);
            }
        }
    
        /**
         * 检测参数是否正确
         *
         * @param openId 用户的openId
         * @param client 客户端类型
         */
        private void checkSocketParam(String openId, String client) {
             log.info("长连接的openId:{}, client{}", openId, client);
            if (!ClientType.checkParamIsHave(client) ||
                StringUtil.isEmpty(openId) || userInfoMapper.getUserInfoByOpenId(openId) == null) {
                throw new SocketException(ErrorCodeConstants.PARAM_ERROR);
            }
    
        }
    
        /**
         * 检查token是否合法
         *
         * @param token 连接时的身份令牌
         */
        private void checkToken(String token) {
            if (StringUtils.isEmpty(token)) {
                log.error("token异常{}", token);
                throw new SocketException(ErrorCodeConstants.PARAM_ERROR);
            }
            /* 验证token 合法性*/
            CommonService commonService = SpringFactory.getObject(CommonService.class);
            Token tk = commonService.getTokenObjByToken(token);
            if (tk == null) {
                throw new SocketException(ErrorCodeConstants.TOKEN_ILLEGAL);
            }
        }
    
        //根据openId和客户端类型获取连接实体
        private synchronized SocketSession getSocketInfo(String openId, String client) {
            SocketSession socketSession = null;
            if (connectionSet.size() == 0) {
                return socketSession;
            }
            for (SocketSession sessionEntity : connectionSet) {
                //连接已存在
                if (sessionEntity.getOpenId().equals(openId) && sessionEntity.getClient().getValue().equals(client)) {
                    socketSession = sessionEntity;
                    break;
                }
            }
            return socketSession;
        }
    
        private static synchronized int getOnlineCount() {
            return connectionSet.size();
        }
    
        /**
         * 维持心跳心跳线程
         */
        private void startHeart() {
            ExamineHeartThread examineHeartThread = new ExamineHeartThread();
            Thread examineThread = new Thread(examineHeartThread);
    
            KeepHeart keepHeart = new KeepHeart();
            Thread keepHeartThread = new Thread(keepHeart);
    
            examineThread.start();
            keepHeartThread.start();
        }
    
        /**
         * 检查是否收到过心跳
         */
        private class ExamineHeartThread implements Runnable {
    
            @Override
            public void run() {
                while (true) {
                    for (SocketSession sessionEntity : connectionSet) {
                        if (!sessionEntity.isHeart() && sessionEntity.getTimeStr() != 0 && sessionEntity.getTimeStr() < getTimeInMillis()) {
                            onClose(sessionEntity.getOpenId(), sessionEntity.getClient().getValue());
                            log.info("openId为{}，客户端为{}的连接未收到过心跳包，断开连接", sessionEntity.getOpenId(), sessionEntity.getClient());
                        }
                    }
                    try {
                        Thread.sleep(50000);
                    } catch (Exception e) {
                        log.info("设置检查心跳的线程睡眠失败" + e);
                    }
                }
            }
        }
    
        private static long getTimeInMillis() {
            long time = System.currentTimeMillis();
            return time - 30000L*5;
        }
    
        /**
         * 保持心跳，定时发送心跳包
         */
        private class KeepHeart implements Runnable {
    
            @Override
            public void run() {
                JSONObject heartJson = new JSONObject();
                heartJson.put("type", "qingtui");
                heartJson.put("secret", "pong");
                while (true) {
                    try {
                        log.info("发送心跳包当前人数为:" + getOnlineCount());
                        sendPing(heartJson.toString());
                        Thread.sleep(30000);
                    } catch (InterruptedException e) {
                        log.info("发送消息包失败" +e);
                    }
                }
            }
        }
    
        /**
         * 发送心跳包
         *
         * @param message 心跳包消息体
         */
        private synchronized void sendPing(String message) {
            if (connectionSet.size() <= 0) {
                return;
            }
            for (SocketSession socketSession : connectionSet) {
                socketSession.setHeart(false);
                try {
                    if (socketSession.getSession().isOpen())
                    socketSession.getSession().getBasicRemote().sendText(message);
                } catch (Exception e) {
                    log.error("发生了异常" + e);
                    onClose(socketSession.getOpenId(), socketSession.getClient().getValue());
                }
            }
    
        }
    }

SocketSession类

/**
     * socket保存连接
     *
     * @author qiaofeng
     */
    @Data
    public class SocketSession {
    
        private String openId;//用户id
        private Session session;
        private ClientType client;//客户端类型
        private long timeStr;//记录下次发送时间的时间戳
        private boolean isHeart=false;//是否收到了心跳
    
        public SocketSession(){}
    
        public SocketSession(String openId, ClientType client, Session session){
            this.openId = openId;
            this.client = client;
            this.session = session;
        }
    
    }

[https_mp.csdn.net_postedit_87882743]: https://mp.csdn.net/postedit/87882743