devbox
你好赵伟
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Java版的WebSocket_javax.websocket

作者：你好赵伟 | 2024-03-16 11:21:46

赞

踩

javax.websocket

项目背景介绍：公司新开发的APP，需要使用系统公告功能，实现实时更新系统公告功能。

 公告的修改功能，由PC端的运营管理界面来进行修改更新发布。当修改完后，要保证APP端，在用户不刷新或重新登录APP的情况下，更新公告内容。

项目的实际架构部署：

负载均衡的后面有两台服务器，部署的服务都是一样的。通过nginx来代理转发请求的具体服务。

多实例部署的情况下，会造成websocket的session不能共享。这里采用的是：redis的发布和订阅功能，实现修改后的公告消息在每台服务器上都能收到，然后再获取每台机器上的websocket会话，给APP端发送消息。 

  1. public class Constants
  2. {
  3.     /** redis 订阅消息通道标识*/
  4.     public final static String REDIS_CHANNEL = "onMessage";
  5.     public final static String REDIS_CHANNEL_CLOSE="close";
  6.     public final static String REDIS_CHANNEL_SEND="send";
  7. }
  1.  @Autowired
  2.  private StringRedisTemplate stringRedisTemplate;
  3.  
  4.  private void sendMessage(String message) {
  5.         String newMessge= null;
  6.         try {
  7.             newMessge = new String(message.getBytes(Constants.UTF8), Constants.UTF8);
  8.         } catch (UnsupportedEncodingException e) {
  9.             e.printStackTrace();
  10.         }
  11. //        Map<String,String> map = new HashMap<String, String>();
  12. //        map.put(Constants.REDIS_MESSAGE_KEY, key);
  13. //        map.put(Constants.REDIS_MESSAGE_VALUE, newMessge);
  14.         stringRedisTemplate.convertAndSend(Constants.REDIS_CHANNEL, newMessge);//发布消息
  15.         System.out.println("将消息: "+newMessge+ "发布出去...");
  16.     }
  1. package cn.com.yuanquanyun.config;
  2.  
  3. import cn.com.yuanquanyun.client.service.ws.WebSocketServer;
  4. import cn.com.yuanquanyun.common.constant.Constants;
  5. import cn.com.yuanquanyun.common.utils.StringUtils;
  6. import com.alibaba.fastjson.JSON;
  7. import com.alibaba.fastjson.JSONObject;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.data.redis.connection.Message;
  12. import org.springframework.data.redis.connection.MessageListener;
  13. import org.springframework.stereotype.Component;
  14.  
  15. import java.io.IOException;
  16. import java.util.Collection;
  17. import java.util.Iterator;
  18. import java.util.concurrent.ConcurrentHashMap;
  19.  
  20.  
  21. /**
  22.  * 消息监听对象，接收订阅消息 -- 每台服务器都订阅公告消息
  23.  */
  24. @Component
  25. public class RedisReceiver implements MessageListener {
  26.     Logger log = LoggerFactory.getLogger(this.getClass());
  27.  
  28.     //@Autowired
  29.     //private WebSocketServer webSocketServer;
  30.  
  31.  
  32.     /**
  33.      * 处理接收到的订阅消息
  34.      */
  35.     @Override
  36.     public void onMessage(Message message, byte[] pattern) {
  37.         String channel = new String(message.getChannel());// 订阅的频道名称
  38.         String msg = "";
  39.         try {
  40.             msg = new String(message.getBody(), Constants.UTF8);//注意与发布消息编码一致，否则会乱码
  41.             System.out.println("打印订阅的消息:" + msg);
  42.             if (!StringUtils.isEmpty(msg)) {
  43.                 if (Constants.REDIS_CHANNEL.endsWith(channel))// 最新消息
  44.                 {
  45.                     //JSONObject jsonObject = JSON.parseObject(msg);
  46.  
  47.                     //将更新后的公告推送给在线的用户手机端
  48.                     ConcurrentHashMap<String, WebSocketServer> webSocketMap = WebSocketServer.getWebSocketMap();
  49.                     Collection<WebSocketServer> values = webSocketMap.values();
  50.                     System.err.println("values ===" + values.size());
  51.                     Iterator<WebSocketServer> it = values.iterator();
  52.                     while (it.hasNext()) {
  53.                         WebSocketServer ws = it.next();
  54.                         try {
  55.                             ws.sendMessage(msg);
  56.                         } catch (IOException e) {
  57.                             System.err.println("sendMessage_error=" + e);
  58.                         }
  59.                     }
  60.  
  61. //                    webSocketServer.sendMessageByWayBillId(
  62. //                            Long.parseLong(jsonObject.get(Constants.REDIS_MESSAGE_KEY).toString())
  63. //                            ,jsonObject.get(Constants.REDIS_MESSAGE_VALUE).toString());
  64.                 } else {
  65.                     //TODO 其他订阅的消息处理
  66.                 }
  67.  
  68.             } else {
  69.                 log.info("消息内容为空，不处理。");
  70.             }
  71.         } catch (Exception e) {
  72.             log.error("处理消息异常：" + e.toString());
  73.             e.printStackTrace();
  74.         }
  75.     }
  76. }
  1. package cn.com.yuanquanyun.config;
  2.  
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.web.socket.config.annotation.EnableWebSocket;
  6. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
  7. /**
  8.  * 开启websocket支持
  9.  * @author Administrator
  10.  */
  11. @Configuration
  12. //@EnableWebSocket
  13. public class WebSocketConfig {
  14. 	
  15. 	@Bean  
  16.     public ServerEndpointExporter serverEndpointExporter() {
  17.         System.err.println("WebSocketConfig 初始化 ");
  18. 	    return new ServerEndpointExporter();
  19.     }
  20. }
<!-- springboot集成websocket -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

  1. package cn.com.yuanquanyun.client.service.ws;
  2.  
  3. import cn.com.yuanquanyun.client.dto.CustNotice;
  4. import cn.com.yuanquanyun.client.service.impl.CustNoticeServiceImpl;
  5. import cn.com.yuanquanyun.common.utils.spring.SpringUtils;
  6. import cn.hutool.log.Log;
  7. import cn.hutool.log.LogFactory;
  8. import org.springframework.stereotype.Component;
  9.  
  10. import javax.websocket.*;
  11. import javax.websocket.server.PathParam;
  12. import javax.websocket.server.ServerEndpoint;
  13. import java.io.IOException;
  14. import java.util.concurrent.ConcurrentHashMap;
  15.  
  16. /**
  17.  * 此方法使用于工程部署是单例的情况下，若是多例的情况下，存在userId不在一台机器上，webSocketMap中查询不到。
  18.  * 多机的情况下，要考虑数据共享的问题，可采用redis发布和订阅来解决.
  19.  * @author Administrator
  20.  *
  21.  */
  22. @Component
  23. @ServerEndpoint("/ws/{userId}")
  24. public class WebSocketServer {
  25. 	static Log log=LogFactory.get(WebSocketServer.class);
  26.     /**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/
  27. //    private static int onlineCount = 0;
  28.     /**concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。*/
  29.     private static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
  30.     /**与某个客户端的连接会话，需要通过它来给客户端发送数据*/
  31.     private Session session;
  32.     /**接收userId*/
  33.     private String userId="";
  34.  
  35.     public WebSocketServer() {
  36. 		System.out.println("WebSocketServer 初始化... hashCode="+this.hashCode());
  37. 	}
  38.     public static ConcurrentHashMap<String,WebSocketServer> getWebSocketMap(){
  39.         return webSocketMap;
  40.     }
  41.     
  42.     /**
  43.      * 连接建立成功调用的方法 只调一次
  44.      *
  45.      */
  46.     @OnOpen
  47.     public void onOpen(Session session,@PathParam("userId") String userId) {
  48.     	System.out.println("onOpen hashCode="+this.hashCode()+", userId="+userId);//根据打印的hashCode每建立一个连接就会生成一个新的WebSocketServer对象，与对应的userId对应
  49.         this.session = session;
  50.         this.userId=userId;
  51.         if(webSocketMap.containsKey(userId)){
  52.             webSocketMap.remove(userId);
  53.             webSocketMap.put(userId,this);
  54.             //加入set中
  55.         }else{
  56.             webSocketMap.put(userId,this);
  57.             //加入set中
  58.             //addOnlineCount();
  59.             //在线数加1
  60.         }
  61.  
  62.         //log.info("用户连接:"+userId+",当前在线人数为:" + getOnlineCount());
  63.  
  64.         try {
  65. //            System.out.println("custNoticeService======"+SpringUtils.getBean(CustNoticeServiceImpl.class));
  66.             CustNotice custNotice = SpringUtils.getBean(CustNoticeServiceImpl.class).selectNotice();
  67. //            CustNotice custNotice = custNoticeService.selectNotice();
  68. //            System.out.println("custNotice======"+custNotice.toString());
  69.             sendMessage(custNotice.getNoticeContent());
  70.         } catch (IOException e) {
  71.             log.error("用户:"+userId+",网络异常!!!!!!");
  72.         }
  73.     }
  74.     
  75.     /**
  76.      * 收到客户端消息后调用的方法
  77.      *
  78.      * @param message 客户端发送过来的消息*/
  79.     @OnMessage
  80.     public void onMessage(String message, Session session) {
  81.     	/*System.out.println("onMessage hashcode=="+this.hashCode());
  82.         log.info("onMessage 用户消息:"+userId+",报文:"+message);
  83.         //可以群发消息
  84.         //消息保存到数据库、redis
  85.         if(StringUtils.isNotBlank(message)){
  86.             try {
  87.                 //解析发送的报文
  88.                 JSONObject jsonObject = JSON.parseObject(message);
  89.                 //追加发送人(防止串改)
  90.                 jsonObject.put("fromUserId",this.userId);
  91.                 String toUserId=jsonObject.getString("toUserId");
  92.                 //传送给对应toUserId用户的websocket
  93.                 if(StringUtils.isNotBlank(toUserId)&&webSocketMap.containsKey(toUserId)){
  94.                     webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());
  95.                 }else{
  96.                     log.error("请求的userId:"+toUserId+"不在该服务器上");
  97.                     //否则不在这个服务器上，发送到mysql或者redis
  98.                 }
  99.             }catch (Exception e){
  100.                 e.printStackTrace();
  101.             }
  102.         }*/
  103.     }
  104.     
  105.     /**
  106.      * 页面socket连接关闭，自动调用的方法
  107.      */
  108.     @OnClose
  109.     public void onClose() {
  110.         if(webSocketMap.containsKey(userId)){
  111.             webSocketMap.remove(userId);
  112.             //从set中删除
  113. //            subOnlineCount();
  114.         }
  115. //        log.info("用户退出:"+userId+",当前在线人数为:" + getOnlineCount());
  116.     }
  117.     
  118.     /**
  119.     *
  120.     * @param session
  121.     * @param error
  122.     */
  123.    @OnError
  124.    public void onError(Session session, Throwable error) {
  125.        log.error("用户错误:"+this.userId+",原因:"+error.getMessage());
  126.        error.printStackTrace();
  127.    }
  128.    
  129.    /**
  130.     * 实现服务器主动推送
  131.     */
  132.    public void sendMessage(String message) throws IOException {
  133.        this.session.getBasicRemote().sendText(message);
  134.    }
  135.    
  136.    /**
  137.     * 发送自定义消息
  138.    */
  139.    /*public static void sendInfo(String message,@PathParam("userId") String userId) throws IOException {
  140.        log.info("发送消息到:"+userId+"，报文:"+message);
  141.        if(StringUtils.isNotBlank(userId)&&webSocketMap.containsKey(userId)){
  142.            webSocketMap.get(userId).sendMessage(message);
  143.        }else{
  144.            log.error("用户"+userId+",不在线！");
  145.        }
  146.    }*/
  147.  
  148. }
  1. <html>
  2. <head>
  3.     <meta charset="utf-8">
  4.     <title>websocket通讯</title>
  5. </head>
  6. <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.js"></script>
  7. <script>
  8.     var socket;
  9.  
  10.     function openSocket() {
  11.         if (typeof (WebSocket) == "undefined") {
  12.             console.log("您的浏览器不支持WebSocket");
  13.         } else {
  14.             console.log("您的浏览器支持WebSocket");
  15.             //实现化WebSocket对象，指定要连接的服务器地址与端口  建立连接
  16.             //等同于socket = new WebSocket("ws://localhost:8888/xxxx/im/25");
  17.             //var socketUrl="${request.contextPath}/im/"+$("#userId").val();
  18.             // var socketUrl = "https://localhost:9093/websocket/" + $("#userId").val();
  19.             // var socketUrl = "wss://localhost:8096/v3/ws/" + $("#userId").val();
  20.             // var socketUrl = "wss://yu_ming:8096/v3/ws/" + $("#userId").val();
  21.             var socketUrl = "wss://yu_ming:8093/websk/ws/" + $("#userId").val();
  22.             socketUrl = socketUrl.replace("https", "wss").replace("http", "ws");
  23.             console.log(socketUrl);
  24.             if (socket != null) {
  25.                 socket.close();
  26.                 socket = null;
  27.             }
  28.             socket = new WebSocket(socketUrl);
  29.             //打开事件
  30.             socket.onopen = function () {
  31.                 console.log("websocket已打开");
  32.                 //socket.send("这是来自客户端的消息" + location.href + new Date());
  33.             };
  34.             //获得消息事件
  35.             socket.onmessage = function (msg) {
  36.                 console.log(msg.data);
  37.                 //发现消息进入    开始处理前端触发逻辑
  38.             };
  39.             //关闭事件
  40.             socket.onclose = function () {
  41.                 console.log("websocket已关闭");
  42.             };
  43.             //发生了错误事件
  44.             socket.onerror = function () {
  45.                 console.log("websocket发生了错误");
  46.             }
  47.         }
  48.     }
  49.  
  50.     function sendMessage() {
  51.         if (typeof (WebSocket) == "undefined") {
  52.             console.log("您的浏览器不支持WebSocket");
  53.         } else {
  54.             console.log("您的浏览器支持WebSocket");
  55.             console.log($("#contentText").val());
  56.             socket.send($("#contentText").val());
  57.         }
  58.     }
  59. </script>
  60. <body>
  61. <p>【userId】：
  62. <div><input id="userId" name="userId" type="text" value="10"></div>
  63. <p>【toUserId】：
  64. <div><input id="toUserId" name="toUserId" type="text" value="20"></div>
  65. <p>【toUserId】：
  66. <div><input id="contentText" name="contentText" type="text" value="hello websocket"></div>
  67. <p>【操作】：
  68. <div><a onclick="openSocket()">开启socket</a></div>
  69. <p>【操作】：
  70. <div><a onclick="sendMessage()">发送消息</a></div>
  71. </body>
  72.  
  73. </html>

通过html界面功能可以模拟和后端建立websocket长连接。使用wss时app端要通过域名来访问。

这里有个小插曲：

生产环境app端请求后台是通过nginx代理转发到服务端的。测试环境没有通过nginx转发。

这里的nginx下的nginx.conf文件里需要配置一下支持websocket协议。

  1. #user  nobody;
  2. worker_processes  1;
  3.  
  4. #error_log  logs/error.log;
  5. #error_log  logs/error.log  notice;
  6. #error_log  logs/error.log  info;
  7.  
  8. #pid        logs/nginx.pid;
  9.  
  10.  
  11. events {
  12.     worker_connections  1024;
  13. }
  14.  
  15.  
  16. http {
  17.     include       mime.types;
  18.     default_type  application/octet-stream;
  19.  
  20.     #log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
  21.     #                  '$status $body_bytes_sent "$http_referer" '
  22.     #                  '"$http_user_agent" "$http_x_forwarded_for"';
  23.  
  24.     #access_log  logs/access.log  main;
  25.  
  26.     sendfile        on;
  27.     #tcp_nopush     on;
  28.  
  29.     #keepalive_timeout  0;
  30.     keepalive_timeout  65;
  31.  
  32.     #gzip  on;
  33.     map $http_upgrade $connection_upgrade {
  34.     default upgrade;
  35.     ''      close;
  36.     }
  37.     server {
  38.         listen       8093;
  39.         server_name  localhost;
  40.  
  41.         #charset koi8-r;
  42.  
  43.         #access_log  logs/host.access.log  main;
  44.  
  45.  
  46.         location /v2 {
  47.             proxy_pass  http://10.0.0.10:8082/v2;
  48.         }
  49.  
  50. 	location /v3 {
  51.             proxy_pass  http://10.0.0.10:8083/v3;
  52.         }
  53.  
  54. 	location /websk {
  55.             proxy_pass http://10.0.0.10:8083/v3;
  56.             
  57.             proxy_read_timeout 300s;
  58.             proxy_send_timeout 300s;
  59.             
  60.             proxy_set_header Host $host;
  61.             proxy_set_header X-real-ip $remote_addr;
  62.             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
  63.             
  64.             proxy_http_version 1.1;
  65.             proxy_set_header Upgrade $http_upgrade;
  66.             proxy_set_header Connection $connection_upgrade;
  67.         }
  68.  
  69.         location /404 {
  70.                 index index.html index.htm;
  71.                 root  /usr/share/nginx/html;
  72.                 error_page 404 /index.html;
  73.         }
  74.         location / {
  75.                 index index.html index.htm;
  76.                 root  /usr/share/nginx/html;
  77.                 error_page 404 /index.html;
  78.         }
  79.         # redirect server error pages to the static page /50x.html
  80.         #
  81.         #error_page   500 502 503 504  /50x.html;
  82.         location = /50x.html {
  83.             root   html;
  84.         }
  85.  
  86.         # proxy the PHP scripts to Apache listening on 127.0.0.1:80
  87.         #
  88.         #location ~ \.php$ {
  89.         #    proxy_pass   http://127.0.0.1;
  90.         #}
  91.  
  92.         # pass the PHP scripts to FastCGI server listening on 127.0.0.1:9000
  93.         #
  94.         #location ~ \.php$ {
  95.         #    root           html;
  96.         #    fastcgi_pass   127.0.0.1:9000;
  97.         #    fastcgi_index  index.php;
  98.         #    fastcgi_param  SCRIPT_FILENAME  /scripts$fastcgi_script_name;
  99.         #    include        fastcgi_params;
  100.         #}
  101.  
  102.         # deny access to .htaccess files, if Apache's document root
  103.         # concurs with nginx's one
  104.         #
  105.         #location ~ /\.ht {
  106.         #    deny  all;
  107.         #}
  108.     }
  109.  
  110.  
  111.     # another virtual host using mix of IP-, name-, and port-based configuration
  112.     #
  113.     #server {
  114.     #    listen       8000;
  115.     #    listen       somename:8080;
  116.     #    server_name  somename  alias  another.alias;
  117.  
  118.     #    location / {
  119.     #        root   html;
  120.     #        index  index.html index.htm;
  121.     #    }
  122.     #}
  123.  
  124.  
  125.     # HTTPS server
  126.     #
  127.     #server {
  128.     #    listen       443 ssl;
  129.     #    server_name  localhost;
  130.  
  131.     #    ssl_certificate      cert.pem;
  132.     #    ssl_certificate_key  cert.key;
  133.  
  134.     #    ssl_session_cache    shared:SSL:1m;
  135.     #    ssl_session_timeout  5m;
  136.  
  137.     #    ssl_ciphers  HIGH:!aNULL:!MD5;
  138.     #    ssl_prefer_server_ciphers  on;
  139.  
  140.     #    location / {
  141.     #        root   html;
  142.     #        index  index.html index.htm;
  143.     #    }
  144.     #}
  145.  
  146. }

里面的关键部分是：

  1. map $http_upgrade $connection_upgrade {
  2.     default upgrade;
  3.     ''      close;
  4. }
  5.  
  6. location /websk {
  7.             proxy_pass http://10.0.0.10:8083/v3;
  8.             
  9.             proxy_read_timeout 300s;
  10.             proxy_send_timeout 300s;
  11.             
  12.             proxy_set_header Host $host;
  13.             proxy_set_header X-real-ip $remote_addr;
  14.             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
  15.             
  16.             proxy_http_version 1.1;
  17.             proxy_set_header Upgrade $http_upgrade;
  18.             proxy_set_header Connection $connection_upgrade;
  19. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/249154
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号