聊天室信息发送

pom.xml

@@ -103,6 +103,10 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-aop</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/com/zhangshu/chat/demo/config/WebSocketConfig.java

@@ -0,0 +1,18 @@
+package com.zhangshu.chat.demo.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.server.standard.ServerEndpointExporter;
+
+@Configuration
+public class WebSocketConfig {
+    /**
+     * 注入ServerEndpointExporter，
+     * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
+     */
+    @Bean
+    public ServerEndpointExporter serverEndpointExporter() {
+        return new ServerEndpointExporter();
+    }
+
+}

src/main/java/com/zhangshu/chat/demo/constant/ERoomUserType.java

@@ -1,6 +1,7 @@
 package com.zhangshu.chat.demo.constant;
 
 public enum ERoomUserType {
+    lord(),
     broadcaster(),
     audience(),
     ;

src/main/java/com/zhangshu/chat/demo/service/EventServiceImpl.java

@@ -56,6 +56,7 @@ public class EventServiceImpl implements EventService {
                         .username(user.getUsername())
                         .nickname(user.getNickname())
                         .type(ERoomUserType.broadcaster)
+                        .joinTime(System.currentTimeMillis())
                         .lastClientSeq(eventDto.getClientSeq())
                         .build());
             }
@@ -77,6 +78,7 @@ public class EventServiceImpl implements EventService {
                         .id(eventDto.getUid())
                         .nickname(user.getNickname())
                         .type(ERoomUserType.audience)
+                        .joinTime(System.currentTimeMillis())
                         .lastClientSeq(eventDto.getClientSeq())
                         .build());
             }

src/main/java/com/zhangshu/chat/demo/service/RoomCache.java

@@ -6,15 +6,18 @@ import cn.hutool.core.lang.UUID;
 import com.zhangshu.chat.demo.constant.ERoomUserType;
 import com.zhangshu.chat.demo.entity.Room;
 import com.zhangshu.chat.demo.vo.RoomUserVo;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Objects;
 
 @Component
 public class RoomCache {
     private final Cache<String, Room> roomCache = CacheUtil.newFIFOCache(4);
+    private final Cache<Long, String> userRoomCache = CacheUtil.newFIFOCache(16);
     /**
      * 正在创建
      */
@@ -26,6 +29,24 @@ public class RoomCache {
         return resp;
     }
 
+    /**
+     *
+     * @param userId
+     * @return
+     *  NullException
+     */
+    public Room getRoomByUser(Long userId){
+        String roomId = userRoomCache.get(userId);
+        if (StringUtils.isBlank(roomId)) {
+            return null;
+        }
+        return roomCache.get(roomId);
+    }
+
+    public boolean isExistUser(Long userId){
+        return userRoomCache.containsKey(userId);
+    }
+
     public synchronized Room addCreating(String roomName) {
         Room room = Room.builder()
                 .id(UUID.fastUUID().toString())
@@ -55,7 +76,11 @@ public class RoomCache {
     public synchronized void addUser(String roomId, RoomUserVo user) {
         Room room = roomCache.get(roomId);
         if (Objects.nonNull(room) && room.getUserList().stream().noneMatch(v -> v.getId().equals(user.getId()))) {
+            if (room.getUserList().size() == 0) {
+                user.setType(ERoomUserType.lord);
+            }
             room.getUserList().add(user);
+            userRoomCache.put(user.getId(), room.getId());
         }
     }
 
@@ -66,7 +91,16 @@ public class RoomCache {
     public synchronized void removeUser(String roomId, Long userId) {
         Room room = roomCache.get(roomId);
         if (Objects.nonNull(room)) {
-            room.getUserList().stream().filter(v -> v.getId().equals(userId)).findFirst().ifPresent(v -> room.getUserList().remove(v));
+            room.getUserList().stream().filter(v -> v.getId().equals(userId)).findFirst().ifPresent(user -> {
+                room.getUserList().remove(user);
+                userRoomCache.remove(userId);
+                //设置最早的一个成为房主
+                if (ERoomUserType.lord.equals(user.getType())) {
+                    room.getUserList().stream().filter(v->ERoomUserType.broadcaster.equals(v.getType()))
+                            .min(Comparator.comparing(RoomUserVo::getJoinTime))
+                            .ifPresent(user2 -> user2.setType(ERoomUserType.lord));
+                }
+            });
         }
     }
 

src/main/java/com/zhangshu/chat/demo/service/RoomServiceImpl.java

@@ -11,6 +11,9 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.List;
+import java.util.stream.Collectors;
+
 @Slf4j
 @Service
 public class RoomServiceImpl implements RoomService {
@@ -20,7 +23,14 @@ public class RoomServiceImpl implements RoomService {
 
     @Override
     public CommonPageResult<RoomVo> page(PageDto pageDto) {
-        return CommonPageResult.of(RoomConvert.INSTANCE.convert(roomCache.list()));
+        List<RoomVo> roomVoList = roomCache.list().stream().map(this::buildRoomVo).collect(Collectors.toList());
+        return CommonPageResult.of(roomVoList);
+    }
+
+    private RoomVo buildRoomVo(Room room){
+        RoomVo roomVo = RoomConvert.INSTANCE.convert(room);
+        roomVo.setUserCount(room.getUserList().size());
+        return roomVo;
     }
 
     @Override

src/main/java/com/zhangshu/chat/demo/vo/RoomUserVo.java

@@ -16,4 +16,5 @@ public class RoomUserVo {
     private String nickname;
     private ERoomUserType type;
     private Long lastClientSeq;
+    private Long joinTime;
 }

src/main/java/com/zhangshu/chat/demo/vo/RoomVo.java

@@ -6,4 +6,5 @@ import lombok.Data;
 public class RoomVo {
     private String id;
     private String name;
+    private Integer userCount;
 }

src/main/java/com/zhangshu/chat/demo/vo/UserMessageVo.java

@@ -0,0 +1,31 @@
+package com.zhangshu.chat.demo.vo;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@ApiModel(value = "UserMessageVo", description = "用户发送的聊天消息")
+public class UserMessageVo implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    @ApiModelProperty(name = "id", value = "user id")
+    private Long id;
+
+    @ApiModelProperty(name = "username", value = "用户名")
+    private String username;
+
+    @ApiModelProperty(name = "nickname", value = "昵称")
+    private String nickname;
+
+    @ApiModelProperty(name = "message", value = "消息内容")
+    private String message;
+}

src/main/java/com/zhangshu/chat/demo/websocket/ChatRoom.java

@@ -0,0 +1,139 @@
+package com.zhangshu.chat.demo.websocket;
+
+import cn.hutool.cache.Cache;
+import cn.hutool.cache.CacheUtil;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.zhangshu.chat.demo.entity.Room;
+import com.zhangshu.chat.demo.service.RoomCache;
+import com.zhangshu.chat.demo.vo.RoomUserVo;
+import com.zhangshu.chat.demo.vo.UserMessageVo;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.websocket.*;
+import javax.websocket.server.PathParam;
+import javax.websocket.server.ServerEndpoint;
+import java.io.IOException;
+import java.util.Objects;
+import java.util.Optional;
+
+@Slf4j
+@Component
+@ServerEndpoint("/chat/room/{userId}")
+public class ChatRoom {
+    private final Cache<String, Long> sessionUserCache = CacheUtil.newFIFOCache(16);
+    private final Cache<String, Session> sessionCache = CacheUtil.newFIFOCache(16);
+    private final Cache<Long, String> userSessionCache = CacheUtil.newFIFOCache(16);
+    @Autowired
+    RoomCache roomCache;
+    @Autowired
+    ObjectMapper objectMapper;
+
+    /**
+     * 建立连接
+     *
+     * @param session 客户端连接对象
+     */
+    @OnOpen
+    public void onOpen(Session session, @PathParam("userId") Long userId) {
+        if (roomCache.isExistUser(userId)) {
+            sendMessage("用户不在房间中，拒绝连接", session);
+            return;
+        }
+        String sessionId = session.getId().toLowerCase();
+        sessionUserCache.put(sessionId, userId);
+        sessionCache.put(sessionId, session);
+        userSessionCache.put(userId, sessionId);
+        log.info("客户端连接建立成功，User ID：{}，当前在线数：{}", userId, sessionUserCache.size());
+    }
+
+    /**
+     * 接收客户端消息
+     *
+     * @param message 客户端发送的消息内容
+     * @param session 客户端连接对象
+     */
+    @OnMessage
+    @SneakyThrows
+    public void onMessage(String message, Session session) {
+        log.info("服务端接收消息成功，消息内容：{}", message);
+        // 处理消息，并响应给客户端
+        String sessionId = session.getId().toLowerCase();
+        Long userId = sessionUserCache.get(sessionId);
+        if (Objects.isNull(userId)) {
+            return;
+        }
+        Room room = roomCache.getRoomByUser(userId);
+        if (Objects.isNull(room)) {
+            return;
+        }
+        Optional<RoomUserVo> userVo = room.getUserList().stream().filter(v -> v.getId().equals(userId)).findFirst();
+        if (!userVo.isPresent()) {
+            return;
+        }
+        UserMessageVo messageVo = UserMessageVo.builder()
+                .id(userId)
+                .nickname(userVo.get().getNickname())
+                .username(userVo.get().getUsername())
+                .message(message)
+                .build();
+        String text = objectMapper.writeValueAsString(messageVo);
+        room.getUserList().stream()
+                .map(RoomUserVo::getId)
+                .map(userSessionCache::get)
+                .map(sessionCache::get)
+                .forEach(v -> {
+                    this.sendMessage(text, v);
+                });
+        log.info("服务端响应消息成功，接收的User ID：{}，响应内容：{}", userId, text);
+    }
+
+    /**
+     * 处理消息，并响应给客户端
+     *
+     * @param message 发送的消息内容
+     * @param session 发送对象
+     */
+
+    private void sendMessage(String message, Session session) {
+        try {
+            session.getBasicRemote().sendText(message);
+        } catch (IOException e) {
+            log.error("服务端响应消息异常：{}", e.getMessage());
+        }
+
+    }
+
+    /**
+     * 关闭连接
+     *
+     * @param session 客户端连接对象
+     */
+    @OnClose
+    public void onClose(Session session) {
+        String sessionId = session.getId().toLowerCase();
+        Long userId = sessionUserCache.get(sessionId);
+        sessionUserCache.remove(sessionId);
+        sessionCache.remove(sessionId);
+        userSessionCache.remove(userId);
+        log.info("客户端连接关闭成功，User ID：{}，当前在线数：{}", userId, sessionUserCache.size());
+    }
+
+    /**
+     * 连接异常
+     *
+     * @param session 客户端连接对象
+     * @param error   异常
+     */
+    @OnError
+    public void onError(Session session, Throwable error) {
+        String sessionId = session.getId().toLowerCase();
+        Long userId = sessionUserCache.get(sessionId);
+        sessionUserCache.remove(sessionId);
+        sessionCache.remove(sessionId);
+        userSessionCache.remove(userId);
+        log.info("连接异常，User ID：{},error:{}", userId, error);
+    }
+}