分布式springboot+websocket 发送消息

古城微笑少年丶 2023-02-13 11:23 12阅读 0赞

集群版本的websocket，不能单独使用，需要和redis来做消息分发。至于原因，我觉得大家都懂了，bb再多，不如代码一行。

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

package com.cjkj.message.config;
    
    import com.cjkj.common.redis.template.StringRedisUtil;
    import net.sf.json.JSONObject;
    import org.apache.commons.lang.StringUtils;
    import org.apache.log4j.Logger;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import org.springframework.util.CollectionUtils;
    import org.springframework.web.socket.*;
    
    import java.io.IOException;
    import java.util.*;
    
    /**
     * @program: cjkj
     * @description:
     * @author: Mr.Wang
     * @create: 2020-05-14 16:50
     **/
    @Service
    public class CTIHandler implements WebSocketHandler, RedisMsg {
        @Autowired
        private StringRedisUtil stringRedisUtil;
        /**
         * 配置日志
         */
        private static Logger logger = Logger.getLogger(CTIHandler.class);
        /**
         * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
         */
        private static Map<String, List<WebSocketSession>> socketMap = new HashMap<String, List<WebSocketSession>>();
        //新增socket
        @Override
        public void afterConnectionEstablished(WebSocketSession session) throws Exception {
            logger.info("成功建立连接");
            //获取用户信息
            String userName = (String) session.getAttributes().get("userName");
            logger.info("获取当前用户信息:"+userName);
            List<WebSocketSession> sessionList = socketMap.get(userName);
            if (CollectionUtils.isEmpty(sessionList)) {
                List list=new ArrayList();
                list.add(session);
                socketMap.put(userName,list);
            }else {
                socketMap.get(userName).add(session);
            }
            sendMessageToUser(userName, new TextMessage("ok"));
    //        if(socketMap.get(userName)==null) {
    //            List list=new ArrayList();
    //            list.add(session);
    //            socketMap.put(userName,list);
    //            sendMessageToUser(userName, new TextMessage("ok"));
    //            //并且通过redis发布和订阅广播给其他的的机器，或者通过消息队列
    //        }
            logger.info("链接成功");
        }
    
        //接收socket信息
        @Override
        public void handleMessage(WebSocketSession webSocketSession, WebSocketMessage<?> webSocketMessage) throws Exception {
            logger.info("收到信息："+webSocketMessage.toString());
            String userName = (String) webSocketSession.getAttributes().get("userName");
            String payload = webSocketMessage.getPayload().toString();
    
            if (StringUtils.isNotBlank(payload) && "ping".equals(payload)){
                sendMessageToUser(userName, new TextMessage("pong"));
            }
    //        webSocketSession.sendMessage(new TextMessage("aaa"));
    //        sendMessageToUser(userName, new TextMessage("我收到你的信息了"));
        }
    
        /**
         * 发送信息给指定用户
         * @param clientId
         * @param message
         * @return
         */
        public boolean sendMessageToUser(String clientId, TextMessage message) {
            logger.info("clientId:" + clientId);
            WebSocketSession session = socketMap.get(clientId).get(socketMap.get(clientId).size()-1);
            if(session==null) {
                return false;
            }
            logger.info("进入发送消息");
            if (!session.isOpen()) {
                return false;
            }
            try {
                logger.info("正在发送消息");
                session.sendMessage(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
            return true;
        }
    
        public boolean sendMessageToUserList(String clientId, TextMessage message) {
            logger.info("clientId:" + clientId);
            List<WebSocketSession> sessionList = socketMap.get(clientId);
            if(CollectionUtils.isEmpty(sessionList)) {
                return false;
            }
            sessionList.forEach(session -> {
                logger.info("进入发送消息");
                if (!session.isOpen()) {
    
                }
                try {
                    logger.info("正在发送消息");
                    session.sendMessage(message);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
            return true;
        }
    
    
    
        @Override
        public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
            if (session.isOpen()) {
                session.close();
            }
            logger.info("连接出错");
        }
    
        @Override
        public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
            //获取用户信息
            String userName = (String) session.getAttributes().get("userName");
            if(socketMap.get(userName)!=null) {
                //socketMap.remove(userName);
                ListIterator<WebSocketSession> listIter = socketMap.get(userName).listIterator();
                while (listIter.hasNext()) {
                    WebSocketSession s = listIter.next();
                    if (s.getId().equals(session.getId())) {
                        listIter.remove();
                    }
                }
    
                //并且通过redis发布和订阅广播给其他的的机器，或者通过消息队列
            }
            logger.info("连接已关闭：" + status);
        }
    
        @Override
        public boolean supportsPartialMessages() {
            return false;
        }
        /**
         * 接受订阅信息
         */
        @Override
        public void receiveMessage(String message) {
            // TODO Auto-generated method stub
            JSONObject sendMsg = JSONObject.fromObject(message.substring(message.indexOf("{")));
            String clientId = sendMsg.getString("userName");
            TextMessage receiveMessage = new TextMessage(sendMsg.getString("message"));
            boolean flag = sendMessageToUserList(clientId, receiveMessage);
            if(flag) {
                logger.info("我发送消息成功了！");
            }
        }
    
    
    
    }

package com.cjkj.message.config;
    
    import org.springframework.stereotype.Component;
    
    @Component
    public interface RedisMsg {
    
        /**
         * 接受信息
         * @param message
         */
        public void receiveMessage(String message);
    }

package com.cjkj.message.config;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.data.redis.listener.PatternTopic;
    import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
    
    /**
     * @program: cjkj
     * @description:
     * @author: Mr.Wang
     * @create: 2020-05-18 12:49
     **/
    @Configuration
    public class RedisPublishConfig {
    
        /**
         * redis消息监听器容器 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
         * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
         *
         * @param connectionFactory
         * @param listenerAdapter
         * @return
         */
        @Bean // 相当于xml中的bean
        RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                                MessageListenerAdapter listenerAdapter) {
            RedisMessageListenerContainer container = new RedisMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            // 订阅了一个叫chat 的通道
            container.addMessageListener(listenerAdapter, new PatternTopic("websocket"));
            // 这个container 可以添加多个 messageListener
            return container;
        }
    
        /**
         * 消息监听器适配器，绑定消息处理器，利用反射技术调用消息处理器的业务方法
         *
         * @param receiver
         * @return
         */
        @Bean
        MessageListenerAdapter listenerAdapter(RedisMsg receiver) {
            // 这个地方 是给messageListenerAdapter 传入一个消息接受的处理器，利用反射的方法调用“receiveMessage”
            // 也有好几个重载方法，这边默认调用处理器的方法 叫handleMessage 可以自己到源码里面看
            return new MessageListenerAdapter(receiver, "receiveMessage");
        }
    
    }

package com.cjkj.message.config;
    
    import org.springframework.context.annotation.Configuration;
    import org.springframework.web.socket.config.annotation.EnableWebSocket;
    import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
    import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
    
    /**
     * @program: cjkj
     * @description:
     * @author: Mr.Wang
     * @create: 2020-05-14 20:35
     **/
    @Configuration
    @EnableWebSocket
    public class WebSocketConfig implements WebSocketConfigurer {
    
        @Override
        public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
    
            //handler是webSocket的核心，配置入口
            registry.addHandler(new CTIHandler(), "/ws/websocket/{ID}").setAllowedOrigins("*").addInterceptors(new WebSocketInterceptor());
    
        }
    
    }

package com.cjkj.message.config;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.log4j.Logger;
    import org.springframework.http.server.ServerHttpRequest;
    import org.springframework.http.server.ServerHttpResponse;
    import org.springframework.web.socket.WebSocketHandler;
    import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor;
    
    import java.util.Map;
    
    /**
     * @program: cjkj
     * @description:
     * @author: Mr.Wang
     * @create: 2020-05-14 16:47
     **/
    @Slf4j
    public class WebSocketInterceptor extends HttpSessionHandshakeInterceptor {
    
        /**
         * 配置日志
         */
        private static Logger logger = Logger.getLogger(WebSocketInterceptor.class);
        @Override
        public boolean beforeHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse seHttpResponse,
                                       WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
    //    HttpServletRequest request = ((ServletServerHttpRequest) serverHttpRequest).getServletRequest();
            logger.info("当前连接："+serverHttpRequest.getURI());
            String userName = serverHttpRequest.getURI().toString().split("/websocket/")[1];
            attributes.put("userName", userName);
            logger.info("握手之前");
            //从request里面获取对象，存放attributes
            return super.beforeHandshake(serverHttpRequest, seHttpResponse, wsHandler, attributes);
        }
    
        @Override
        public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
                                   Exception ex) {
            logger.info("握手之后");
            super.afterHandshake(request, response, wsHandler, ex);
        }
    }

package com.cjkj.message.controller.websocket;
    
    import com.cjkj.common.redis.template.StringRedisUtil;
    import com.cjkj.message.config.CTIHandler;
    import net.sf.json.JSONObject;
    import org.apache.log4j.Logger;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import org.springframework.web.socket.TextMessage;
    
    /**
     * @program: cjkj
     * @description:
     * @author: Mr.Wang
     * @create: 2020-05-14 17:13
     **/
    @RestController
    @RequestMapping("/socket")
    public class WebSocketPushMegController {
        private static Logger logger = Logger.getLogger(WebSocketPushMegController.class);
        @Autowired
        private StringRedisUtil stringRedisUtil;
    
        @PostMapping("/sendmessage")
        public String list(String id, String message) {
            JSONObject result = new JSONObject();
            try {
                CTIHandler ctiHandler=new CTIHandler();
                Boolean flag = ctiHandler.sendMessageToUser(id, new TextMessage(message));
                //if (!flag) {//发送失败广播出去，让其他节点发送
                    //广播消息到各个订阅者
                    JSONObject message1 = new JSONObject();
                    message1.put("userName", id);
                    message1.put("message", message);
                    stringRedisUtil.convertAndSend("websocket", message1.toString());
               // }
            } catch (Exception e) {
                e.printStackTrace();
                logger.error("推送给客户端失败");
                result.put("result", "error");
            }
            result.put("result", "success");
            return result.toString();
        }
    
    }

其中redis的工具类可以自己用redistemplate。

registry.addHandler(new CTIHandler(), "/ws/websocket/\{ID\}").setAllowedOrigins("\*").addInterceptors(new WebSocketInterceptor());

这个就是接口访问地址，可以用postman测试

遇到的问题，因为我们项目的网关用的zuul，但是zuul并不支持tcp，所以连接不通。然后我们就绕过网关，用nginx做代理直接访问。如果大家对这个又研究，还请告知具体实现方式。

nginx配置websocket也需要配置，否则访问不通。

upstream wsserver \{    
    server 10.253.96.128:8088;  
    server 10.253.96.127:8088;  
    \#server 10.253.96.122:8080;    
\}

server \{  
        listen       80;  
        server\_name  localhost;  
    client\_max\_body\_size 50M;  
        \#charset koi8-r;

location /ws/\{

proxy\_set\_header X-Forwarded-For $proxy\_add\_x\_forwarded\_for;  
        \#rewrite ^/ws/(.\*)$ /$1 break;  
        proxy\_pass http://wsserver/;  
        proxy\_http\_version 1.1;  
        proxy\_set\_header Upgrade $http\_upgrade;  
        proxy\_set\_header Connection "upgrade";  
    \}