前言

使用WebSocket实现一对一的聊天功能与未读消息功能

一、数据库设计

会话表

字段名	字段类型	长度	注释
conversation_id	int	11	会话ID
create_time	datetime		创建时间
conversation_type	int	1	会话类型

消息表

字段名	字段类型	长度	注释
message_id	int	11	消息ID
conversation_id	int	11	会话ID
sender_id	int	11	发送者ID
receiver_id	in t	11	接收者ID
content	text		消息内容
type	int	2	消息类型
information	varchar	255	信息
sender_img	int	11	发送者头像ID
receiver_img	int	11	接收者头像ID
message_status	int	1	消息状态（1已读，0未读）
create_time	datetime		创建时间

二、实现代码

1.SessionWrap

@Data
public class SessionWrap {private String from;	// 连接人idprivate String type;	// 连接类型private Session session;private Date lastTime;
}

2.websocket

@Component
@ServerEndpoint(value = "/api/websocket/{from}/{type}")
public class WebSocketServer {@Autowiredprivate RqriMessageService rqriMessageService;public static WebSocketServer webSocketServer;// 所有的连接会话private static CopyOnWriteArraySet<SessionWrap> sessionList = new CopyOnWriteArraySet<>();private String from;private String type;@PostConstructpublic void init() {webSocketServer = this;webSocketServer.rqriMessageService = this.rqriMessageService;}@OnOpenpublic void onOpen(Session session, @PathParam(value = "from") String from, @PathParam(value = "type") String type) {this.from = from;this.type = type;try {// 遍历list，如果有会话，更新，如果没有，创建一个新的for (SessionWrap item : sessionList) {if (item.getFrom().equals(from) && item.getType().equals(type)) {item.setSession(session);item.setLastTime(new Date());log.info("【websocket消息】更新连接，总数为:" + sessionList.size());return;}}SessionWrap sessionWrap = new SessionWrap();sessionWrap.setFrom(from);sessionWrap.setType(type);sessionWrap.setSession(session);sessionWrap.setLastTime(new Date());sessionList.add(sessionWrap);log.info("【websocket消息】有新的连接，总数为:" + sessionList.size());} catch (Exception e) {log.info("【websocket消息】连接失败！错误信息：" + e.getMessage());}}@OnClosepublic void onClose() {try {sessionList.removeIf(item -> item.getFrom().equals(from) && item.getType().equals(type));log.info("【websocket消息】连接断开，总数为:" + sessionList.size());} catch (Exception e) {log.info("【websocket消息】连接断开失败！错误信息：" + e.getMessage());}}@OnMessagepublic void onMessage(String message, Session session) {try {if ("ping".equals(message)) {session.getBasicRemote().sendText("ping");	// 心跳检测} else {// 将消息插入到数据库JSONObject r = webSocketServer.rqriMessageService.insertMessage(message);// 成功if (r.getInteger("code") == 200) {JSONObject data = r.getJSONObject("data");String senderId = data.getString("senderId");      // 发送者String receiverId = data.getString("receiverId");   // 接收者for (SessionWrap item : sessionList) {if (senderId.equals(item.getFrom()) || receiverId.equals(item.getFrom()) ) {item.getSession().getBasicRemote().sendText(r.toJSONString());} }log.info("【websocket消息】发送消息:" + r.toJSONString());}}} catch (Exception e) {log.info("【websocket消息】发送消息失败！错误信息：" + e.getMessage());}}@OnErrorpublic void onError(Session session, Throwable error) {log.error("用户错误,原因:"+error.getMessage());error.printStackTrace();}}

3.insertMessage

private final String rqriMessageStr = "rqri_message_unread_";public JSONObject insertMessage(String message) {JSONObject jsonObject = new JSONObject();RqriMessage rqriMessage = JSONObject.parseObject(message, RqriMessage.class);// 把消息添加到数据库int i = rqriMessageMapper.insertSelective(rqriMessage);// 将未读信息添加到redis  添加接收者的未读String conversationId = String.valueOf(rqriMessage.getConversationId());String receiverId = String.valueOf(rqriMessage.getReceiverId());String key = rqriMessageStr + conversationId + "_" + receiverId;if (redisUtils.get(key) == null) {redisUtils.set(key, 1, 0); 	// 设置永不过期} else {redisUtils.incr(key, 1);	// 未读数量添加1}jsonObject.put("code", 200);jsonObject.put("data", rqriMessage);// 发送者的id和未读数量，返回给前端渲染到页面HashMap<String, Integer> map = new HashMap<>();map.put("num", Integer.valueOf(redisUtils.get(key).toString()));map.put("id", rqriMessage.getSenderId());jsonObject.put("isread", map);return jsonObject;
}