SpringBoot（十三）SpringBoot配置webSocket

在PHP版本的博客中，我使用PHP+swoole实现了webscoket即时聊天的功能。

在java版本的博客中，我也想使用webscoket来实现即时聊天的功能，下边是我实现过程的一个记录。

一：在pom.xml中添加记录

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

二：在config目录添加WebSocketConfig.java文件

package com.springbootblog.config;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.web.socket.config.annotation.EnableWebSocket;import org.springframework.web.socket.server.standard.ServerEndpointExporter;/*** @ Description: 开启WebSocket支持*/@Configuration@EnableWebSocketpublic class WebSocketConfig{@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}}

三：解决@ServerEndpoint注解中的@Autowired注解失效的问题

原理是WebSocket是线程安全的，有用户连接时就会创建一个新的端点实例，一个端WebSocket是多对象的，使用的spring却是单例模式。这两者刚好冲突。

所以在@ServerEndpoint注解下不能使用Autowired注解注入对象，那我在webscoket的控制器中也是需要操作redis以及操作数据库的，那我该如何去处理这个问题呢？

我百思不得其解。最后在百度上一位大神的指点下解决了这个问题。

使用从容器中取对象的工具类

在utils目录中添加SpringUtil.java文件

package com.springbootblog.utils;import org.springframework.beans.BeansException;import org.springframework.context.ApplicationContext;import org.springframework.context.ApplicationContextAware;import org.springframework.stereotype.Component;@Componentpublic class SpringUtil implements ApplicationContextAware{private static ApplicationContext applicationContext;@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {SpringUtil.applicationContext = applicationContext;}public ApplicationContext getApplicationContext(){return applicationContext;}public static Object getBean(String beanName){return applicationContext.getBean(beanName);}public static <T> T getBean(Class<T> clazz){try{return (T)applicationContext.getBean(clazz);}catch(Exception e){return null;}}}

四：在SpringBoot中使用webscoket

package com.springbootblog.controller.fontend;import com.alibaba.fastjson.JSONObject;import com.springbootblog.dao.ChatRecordDao;import com.springbootblog.dao.UserDao;import com.springbootblog.pojo.ChatRecord;import com.springbootblog.pojo.User;import com.springbootblog.utils.Function;import com.springbootblog.utils.RedisUtil;import com.springbootblog.utils.SpringUtil;import lombok.extern.slf4j.Slf4j;import org.json.JSONException;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import org.springframework.stereotype.Service;import javax.websocket.*;import javax.websocket.server.PathParam;import javax.websocket.server.ServerEndpoint;import java.io.IOException;import java.util.ArrayList;import java.util.HashMap;import java.util.List;import java.util.Map;import java.util.concurrent.CopyOnWriteArraySet;@Component                              // 组件@Slf4j                                  // 日志依赖@Service@ServerEndpoint("/api/websocket/{sid}") // 配置webscoket访问链接public class WebSocketServer{/*** 注入redis工具类*/// @Autowired// private RedisUtil redisUtil;private RedisUtil redisUtil = SpringUtil.getBean(RedisUtil.class);/*** 自动装载（dao接口注入）*/// @Autowired// private UserDao userDao;private UserDao userDao = SpringUtil.getBean(UserDao.class);/*** 自动装载（dao接口注入）*/// @Autowired// private ChatRecordDao chatDao;private ChatRecordDao chatDao = SpringUtil.getBean(ChatRecordDao.class);//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。public static int onlineCount = 0;//concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<>();//与某个客户端的连接会话，需要通过它来给客户端发送数据private Session session;//接收sidpublic String sid = "";/*** 连接建立成功调用的方法*/@OnOpenpublic void onOpen(Session session, @PathParam("sid") String sid){this.session = session;//加入set中webSocketSet.add(this);this.sid = sid;//在线数加1addOnlineCount();try{sendMessage("conn_success");System.out.println("有新窗口开始监听:" + sid + ",当前在线人数为:" + getOnlineCount());}catch (IOException | EncodeException e){System.out.println("websocket IO Exception");}}/*** 连接关闭调用的方法*/@OnClosepublic void onClose(){webSocketSet.remove(this);  //从set中删除subOnlineCount();           //在线数减1//断开连接情况下，更新主板占用情况为释放System.out.println("释放的sid为："+sid);//这里写你 释放的时候，要处理的业务System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());}/*** 收到客户端消息后调用的方法* @ Param message 客户端发送过来的消息*/@OnMessagepublic void onMessage(String message, Session session) throws JSONException, IOException, EncodeException {this.session = session;System.out.println("收到来自窗口" + sid + "的信息:" + message);// 获取key对应的值String type = Function.stringToJson(message,"type");String from_user_id = Function.stringToJson(message,"from_user_id");String to_user_id = Function.stringToJson(message,"to_user_id");String msg = Function.stringToJson(message,"msg");String fd = Function.stringToJson(message,"fd");// 用户聊天if ((from_user_id.equals("") || to_user_id.equals("")) && type.equals("userchat")){Map<String,Object> result = new HashMap<>(2);result.put("code", -1);result.put("msg", "参数错误！请重新登录~");sendMessage(message);}if(type.equals("ping")){// 心跳// 在线用户列表Map<String,Object> array = getWebscoketUserList(from_user_id, session.getId());ArrayList<Map<String,Object>> userList = (ArrayList<Map<String, Object>>) array.get("userList");Integer totalUnReadNumber = (Integer) array.get("totalUnReadNumber");// System.out.println("totalUnReadNumber:"+totalUnReadNumber);// System.out.println("userList:"+userList);Map<String,Object> result = new HashMap<>(4);result.put("code", 1);result.put("type", type);result.put("list", userList);result.put("totalUnReadNumber", totalUnReadNumber);// System.out.println(JSONObject.toJSONString(result));sendMessage(JSONObject.toJSONString(result));}else if(type.equals("userchat")){// 用户一对一聊天// System.out.println("type:"+type);// System.out.println("from_user_id:"+from_user_id);// System.out.println("to_user_id:"+to_user_id);// System.out.println("msg:"+msg);// System.out.println("fd:"+fd);// 保存聊天记录saveChatRecord(from_user_id,to_user_id,msg);// Map<String,Object> param = new HashMap<>(1);// param.put("answer", msg);Map<String,Object> result = new HashMap<>(2);result.put("response", "");result.put("type", type);sendMessage(JSONObject.toJSONString(result));}session.getId();//群发消息/*for (WebSocketServer item : webSocketSet){try{item.sendMessage(message);}catch (IOException e){e.printStackTrace();}}//*/}public Map<String,Object> saveChatRecord(String from_user_id, String to_user_id,String msg){ChatRecord chatRecord = new ChatRecord();chatRecord.setFrom_user_id(Integer.valueOf(from_user_id));chatRecord.setTo_user_id(Integer.valueOf(to_user_id));chatRecord.setMsg(msg);Integer res = chatDao.addChatRecord(chatRecord);Map<String,Object> result = new HashMap<>(2);result.put("code", 1);result.put("id", res);return result;}/*** 获取webscoket链接的用户列表* @param user_id* @param fd* @return*/public Map<String,Object> getWebscoketUserList(String user_id,String fd){// 存redis ，10分钟过期，防止用户退出之后redis中存储的数据一直存在。// 存储格式：webscoket-用户id : webscoket连接主键// 组装redis-keyString redisKey = "webscoket-" + user_id;String webscoketFD = "";// 首先获取当前key值(webscoket 连接主键)// System.out.println(redisKey);// System.out.println("redisUtil:"+redisUtil.get(redisKey));// System.out.println("fd:"+fd);if(redisUtil.get(redisKey) != null ){webscoketFD = redisUtil.get(redisKey).toString();}// System.out.println("webscoketFD:"+webscoketFD);// 判断当前主键是否redis中记录的主键一致。(就是判断这个玩意redis中有没有)if(!webscoketFD.equals(fd)){// 设置一个有过期时间的keyredisUtil.set(redisKey, fd,60*60*2);// System.out.println("存储之后的redis:"+redisUtil.get(redisKey));}// 获取指定前缀的Key(返回一个数组)List<String> keyList = redisUtil.findKeysByPattern("webscoket-*");// System.out.println(keyList);Integer totalUnReadNumber = 0;ArrayList<Map<String,Object>> array = new ArrayList<>();for (int i = 0; i < keyList.size(); i++){Object key = keyList.get(i);// System.out.println(key);String[] arr = key.toString().split("-");if(arr.length < 2){break;}String to_user_id = arr[1];String value = redisUtil.get(key.toString()).toString();// System.out.println(to_user_id);// System.out.println(value);Map<String,Object> temp = new HashMap<>(5);temp.put("fd", value);// 查未读消息数量Integer unReadNumber = chatDao.getUnReadNumber(user_id,to_user_id);totalUnReadNumber += unReadNumber;// 查用户信息User userinfo = userDao.getUserInfoById(to_user_id);if(userinfo == null){redisUtil.del(key.toString());continue;}if(!userinfo.getFigureurl_wx().equals("")){temp.put("userlogo", userinfo.getFigureurl_wx());}else{temp.put("userlogo", userinfo.getFigureurl_qq());}temp.put("username", userinfo.getNickname());temp.put("id", userinfo.getId());array.add(temp);}Map<String,Object> result = new HashMap<>(2);result.put("userList", array);result.put("totalUnReadNumber", totalUnReadNumber);return result;}/*** @ Param session* @ Param error*/@OnErrorpublic void onError(Session session, Throwable error){// log.error("发生错误");error.printStackTrace();}/*** 实现服务器主动推送*/public void sendMessage(String message) throws IOException, EncodeException{this.session.getBasicRemote().sendText(message);// this.session.getBasicRemote().sendObject(message);}/*** 群发自定义消息*/public static void sendInfo(String message, @PathParam("sid") String sid) throws IOException{// System.out.println("推送消息到窗口" + sid + "，推送内容:" + message);for (WebSocketServer item : webSocketSet){try{//这里可以设定只推送给这个sid的，为null则全部推送if (sid == null){// item.sendMessage(message);}else if (item.sid.equals(sid)){item.sendMessage(message);}}catch (IOException e){continue;} catch (EncodeException e) {throw new RuntimeException(e);}}}public static synchronized int getOnlineCount(){return onlineCount;}public static synchronized void addOnlineCount(){WebSocketServer.onlineCount++;}public static synchronized void subOnlineCount(){WebSocketServer.onlineCount--;}}

以上大概就是我在SpringBoot中对webscoket的基本使用。

有好的建议，请在下方输入你的评论。