wyc

src/main/java/com/zhentao/groups/MongoDB/controller/GroupMessageController.java

@@ -30,17 +30,12 @@ public class GroupMessageController {
     public String groupMessage(@RequestBody Message message)
     {
         GroupMessage groupMessage = new GroupMessage();
-        groupMessage.setGroup_id(message.getGroup_id());
-        groupMessage.setSender_id(message.getSender_id());
+        groupMessage.setGroupId(message.getGroupId());
+        groupMessage.setSenderId(message.getFromUserId());
         groupMessage.setContent(message.getContent());
-        groupMessage.setContent_type(message.getContent_type());
-        groupMessage.setCreated_at(new Date());
+        groupMessage.setType(message.getType());
+        groupMessage.setTimestamp(System.currentTimeMillis());
         mongoTemplate.save(groupMessage);
-
-
-
-
-
         return "success";
     }
 

src/main/java/com/zhentao/groups/MongoDB/pojo/GroupMessage.java

@@ -8,29 +8,40 @@ import org.springframework.data.mongodb.core.mapping.Document;
 
 import java.util.Date;
 
-
 @Slf4j
 @Document(collection = "group_messages")
 @Data
 public class GroupMessage {
-
     @Id
     private String id;
 
-////    消息唯一id
-//    private String message_id;
+    /**
+     * 群ID
+     */
+    private Long groupId;
+
+    /**
+     * 发送者ID
+     */
+    private String senderId;
 
-//  群id
-    private Long group_id;
-//  发送者ID
-    private Long sender_id;
-//  消息内容
+    /**
+     * 消息内容
+     */
     private String content;
-//  消息类型
-    private String content_type;
-//  发送时间
-    private Date created_at;
 
+    /**
+     * 消息类型
+     */
+    private String type;
 
+    /**
+     * 消息时间戳
+     */
+    private Long timestamp;
 
+    /**
+     * 创建时间
+     */
+    private Date createdAt;
 }

src/main/java/com/zhentao/groups/MongoDB/pojo/Message.java

@@ -1,23 +1,42 @@
 package com.zhentao.groups.MongoDB.pojo;
 
 import lombok.Data;
-
 import java.util.Date;
+
 @Data
 public class Message {
+    /**
+     * 消息类型
+     */
+    private String type;
+
+    /**
+     * 发送者ID
+     */
+    private String fromUserId;
 
-//    //    消息唯一id
-//    private String message_id;
+    /**
+     * 接收者ID
+     */
+    private String toUserId;
 
-    //  群id
-    private Long group_id;
-    //  发送者ID
-    private Long sender_id;
-    //  消息内容
+    /**
+     * 群ID（群聊消息时使用）
+     */
+    private Long groupId;
+
+    /**
+     * 消息内容
+     */
     private String content;
-    //  消息类型
-    private String content_type;
-    //  发送时间
-    private Date created_at;
 
+    /**
+     * 消息时间戳
+     */
+    private Long timestamp;
+
+    /**
+     * 消息创建时间
+     */
+    private Date createdAt;
 }

src/main/java/com/zhentao/groups/controller/groupsController.java

@@ -40,11 +40,4 @@ public class groupsController {
         delGroupMembers.setUId(userIdFromToken);
         return groupsService.DelGroupMembers(delGroupMembers);
     }
-
-
-
-
-
-
-
 }

src/main/java/com/zhentao/groups/dto/AddGroupMembers.java

@@ -19,5 +19,5 @@ public class AddGroupMembers  implements Serializable {
 
     private List<Long> userId;
 
-
+    private String groupssId;
 }

src/main/java/com/zhentao/groups/dto/GroupDto.java

@@ -0,0 +1,11 @@
+package com.zhentao.groups.dto;
+
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class GroupDto {
+    private Long groupId;
+    private List<Long> uid;
+}

src/main/java/com/zhentao/groups/pojo/Groupss.java

@@ -5,6 +5,7 @@ import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 import java.io.Serializable;
 import java.util.Date;
+import java.util.List;
 
 import com.fasterxml.jackson.annotation.JsonFormat;
 import lombok.Data;
@@ -69,7 +70,8 @@ public class Groupss implements Serializable {
      * 修改时间
      */
     private Date updatedAt;
-
+    @TableField(exist = false)
+    private List<GroupMembers> list;
     @TableField(exist = false)
     private static final long serialVersionUID = 1L;
 
@@ -134,4 +136,4 @@ public class Groupss implements Serializable {
         sb.append("]");
         return sb.toString();
     }
-}
+}

src/main/java/com/zhentao/groups/service/GroupsService.java

@@ -3,9 +3,14 @@ package com.zhentao.groups.service;
 
 import com.zhentao.groups.dto.AddGroupMembers;
 import com.zhentao.groups.dto.DelGroupMembers;
+import com.zhentao.groups.dto.GroupDto;
 import com.zhentao.groups.pojo.Groupss;
 import com.baomidou.mybatisplus.extension.service.IService;
 import com.zhentao.vo.Result;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+
+import java.util.List;
 
 /**
 * @author lzy
@@ -23,5 +28,6 @@ public interface GroupsService extends IService<Groupss> {
 
 //    删除人数
     Result DelGroupMembers(DelGroupMembers delGroupMembers);
-
+//  查询所有的一个群ID
+    List<GroupDto> getList();
 }

src/main/java/com/zhentao/groups/service/impl/GroupsServiceImpl.java

@@ -6,12 +6,14 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.zhentao.groups.dto.AddGroupMembers;
 import com.zhentao.groups.dto.AddGroupsDto;
 import com.zhentao.groups.dto.DelGroupMembers;
+import com.zhentao.groups.dto.GroupDto;
 import com.zhentao.groups.mapper.GroupMembersMapper;
 import com.zhentao.groups.pojo.GroupMembers;
 import com.zhentao.groups.pojo.Groupss;
 import com.zhentao.groups.service.GroupMembersService;
 import com.zhentao.groups.service.GroupsService;
 import com.zhentao.groups.mapper.GroupsMapper;
+import com.zhentao.information.cache.GroupMemberCache;
 import com.zhentao.shouye.domain.UserShouye;
 import com.zhentao.shouye.mapper.UserShouyeMapper;
 import com.zhentao.shouye.service.UserShouyeService;
@@ -26,6 +28,7 @@ import org.springframework.transaction.annotation.Transactional;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
+import java.util.stream.Collectors;
 
 /**
 * @author lzy
@@ -43,12 +46,21 @@ public class GroupsServiceImpl extends ServiceImpl<GroupsMapper, Groupss>
 
     @Autowired
     private UserShouyeMapper userShouyeMapper;
-
 //  创建群
+    @Autowired
+    private GroupMembersService groupMembersService;
+    @Autowired
+    private UserShouyeService userShouyeService;
+    @Autowired
+    private GroupsMapper groupsMapper;
+
+    @Autowired
+    private GroupMemberCache groupMemberCache;
+
     @Override
     @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class, timeout = 30)
     public Result addGroup(AddGroupsDto addGroupsDto) {
-        Groupss groups=new Groupss();
+        Groupss groups = new Groupss();
         groups.setGroupId(IdUtil.getSnowflake().nextId());
         groups.setName(addGroupsDto.getName());
         groups.setCreatorId(addGroupsDto.getCreatorId());
@@ -61,7 +73,7 @@ public class GroupsServiceImpl extends ServiceImpl<GroupsMapper, Groupss>
         groups.setUpdatedAt(new Date());
         int insert1 = this.baseMapper.insert(groups);
 
-        GroupMembers groupMembers=new GroupMembers();
+        GroupMembers groupMembers = new GroupMembers();
         groupMembers.setId(IdUtil.getSnowflake().nextId());
         groupMembers.setGroupId(groups.getGroupId());
         groupMembers.setUserId(groups.getCreatorId());
@@ -70,6 +82,11 @@ public class GroupsServiceImpl extends ServiceImpl<GroupsMapper, Groupss>
         groupMembers.setJoinTime(new Date());
         int insert = groupMembersMapper.insert(groupMembers);
 
+        // 更新群成员缓存
+        List<Long> memberIds = new ArrayList<>();
+        memberIds.add(groups.getCreatorId());
+        groupMemberCache.updateGroupMembers(groups.getGroupId(), memberIds);
+
         UserShouye userShouye=new UserShouye();
         userShouye.setId(IdUtil.getSnowflake().nextId());
         userShouye.setUid1(addGroupsDto.getCreatorId());
@@ -85,28 +102,19 @@ public class GroupsServiceImpl extends ServiceImpl<GroupsMapper, Groupss>
         return Result.ERR(null,"创建失败");
     }
 
-
-    @Autowired
-    private GroupMembersService groupMembersService;
-    @Autowired
-    private UserShouyeService userShouyeService;
-    @Autowired
-    private GroupsMapper groupsMapper;
 //    添加人数
     @Override
     @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class, timeout = 30)
     public Result AddGroupMembers(AddGroupMembers addGroupMembers) {
-
         List<GroupMembers> data = new ArrayList<>();
-        Groupss groupss = groupsMapper.selectById(addGroupMembers.getGroupId());
-        System.err.println(groupss);
-        List<UserShouye> userShouyes =new ArrayList<>();
-        List<GroupMembers> list = groupMembersMapper.selectList(new LambdaQueryWrapper<GroupMembers>().eq(GroupMembers::getGroupId, addGroupMembers.getGroupId()));
-        if (list.size()!=0){
-            if (groupss.getMaxMembers()>list.size()){
-                System.err.println(addGroupMembers.getUserId());
-                for (Long groupMembers:addGroupMembers.getUserId()){
-                    GroupMembers groupMembers1=new GroupMembers();
+        List<UserShouye> userShouyes = new ArrayList<>();
+        Groupss groupss = this.getById(addGroupMembers.getGroupssId());
+        List<GroupMembers> list = groupMembersService.list(new LambdaQueryWrapper<GroupMembers>().eq(GroupMembers::getGroupId, addGroupMembers.getGroupssId()));
+
+        if (list.size() != 0) {
+            if (groupss.getMaxMembers() > list.size()) {
+                for (Long groupMembers : addGroupMembers.getUserId()) {
+                    GroupMembers groupMembers1 = new GroupMembers();
                     groupMembers1.setId(IdUtil.getSnowflake().nextId());
                     groupMembers1.setGroupId(groupss.getGroupId());
                     groupMembers1.setUserId(groupMembers);
@@ -115,7 +123,7 @@ public class GroupsServiceImpl extends ServiceImpl<GroupsMapper, Groupss>
                     groupMembers1.setJoinTime(new Date());
                     data.add(groupMembers1);
 
-                    UserShouye userShouye=new UserShouye();
+                    UserShouye userShouye = new UserShouye();
                     userShouye.setId(IdUtil.getSnowflake().nextId());
                     userShouye.setUid1(groupMembers);
                     userShouye.setUid2(null);
@@ -126,17 +134,23 @@ public class GroupsServiceImpl extends ServiceImpl<GroupsMapper, Groupss>
                 }
                 boolean b1 = groupMembersService.saveBatch(data);
                 boolean b = userShouyeService.saveBatch(userShouyes);
-                if (b1!=false && b!=false){
-                    return Result.OK(null,"添加成功");
+                if (b1 && b) {
+                    // 更新群成员缓存
+                    List<Long> memberIds = list.stream()
+                        .map(GroupMembers::getUserId)
+                        .collect(Collectors.toList());
+                    memberIds.addAll(addGroupMembers.getUserId());
+                    groupMemberCache.updateGroupMembers(groupss.getGroupId(), memberIds);
+                    return Result.OK(null, "添加成功");
                 }
-            }else {
-                return Result.ERR(null,"群已满");
+            } else {
+                return Result.ERR(null, "群已满");
             }
-        }else {
-            return Result.ERR(null,"群已不存在");
+        } else {
+            return Result.ERR(null, "群已不存在");
         }
 
-        return Result.ERR(null,"添加失败");
+        return Result.ERR(null, "添加失败");
     }
 
 
@@ -161,11 +175,33 @@ public class GroupsServiceImpl extends ServiceImpl<GroupsMapper, Groupss>
                 }
                 groupMembersMapper.deleteById(groupMembers2);
                 userShouyeMapper.deleteById(userShouye);
+
+                // 更新群成员缓存
+                List<GroupMembers> remainingMembers = groupMembersService.list(
+                    new LambdaQueryWrapper<GroupMembers>()
+                        .eq(GroupMembers::getGroupId, delGroupMembers.getGroupssId())
+                );
+                List<Long> memberIds = remainingMembers.stream()
+                    .map(GroupMembers::getUserId)
+                    .collect(Collectors.toList());
+                groupMemberCache.updateGroupMembers(delGroupMembers.getGroupssId(), memberIds);
+
                 return Result.OK(null,"删除成功");
             }else if (groupMembers.getRole()==1){
                 if (groupMembers2.getRole()==0){
                     groupMembersMapper.deleteById(groupMembers2);
                     userShouyeMapper.deleteById(userShouye);
+
+                    // 更新群成员缓存
+                    List<GroupMembers> remainingMembers = groupMembersService.list(
+                        new LambdaQueryWrapper<GroupMembers>()
+                            .eq(GroupMembers::getGroupId, delGroupMembers.getGroupssId())
+                    );
+                    List<Long> memberIds = remainingMembers.stream()
+                        .map(GroupMembers::getUserId)
+                        .collect(Collectors.toList());
+                    groupMemberCache.updateGroupMembers(delGroupMembers.getGroupssId(), memberIds);
+
                     return Result.OK(null,"删除成功");
                 }else {
                     return Result.ERR(null,"权限不足");
@@ -178,34 +214,28 @@ public class GroupsServiceImpl extends ServiceImpl<GroupsMapper, Groupss>
         }
 
     }
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+//查询所有的群ID和群里面成员ID
+    @Override
+    public List<GroupDto> getList() {
+        List<Groupss> list = this.list();
+        List<GroupDto> list3 = new ArrayList<>();
+//        所有的群ID
+        List<Long> collect = list.stream().map(Groupss::getGroupId).collect(Collectors.toList());
+        List<GroupMembers> list1 = groupMembersService.list();
+        for (Long c:collect) {
+            List<Long> list2 = new ArrayList<>();
+            for (GroupMembers m: list1) {
+                if (c.equals(m.getGroupId())){
+                    list2.add(m.getUserId());
+                }
+            }
+            GroupDto groupDto = new GroupDto();
+            groupDto.setGroupId(c);
+            groupDto.setUid(list2);
+            list3.add(groupDto);
+        }
+        return list3;
+    }
 
 
 }

src/main/java/com/zhentao/information/cache/ChannelCache.java

@@ -23,6 +23,8 @@ public class ChannelCache {
      */
     private static final Map<String, ChannelHandlerContext> USER_CHANNEL_MAP = new ConcurrentHashMap<>();
 
+    private final Map<ChannelHandlerContext, String> reverseMap = new ConcurrentHashMap<>();
+
     /**
      * 添加用户Channel映射
      * @param userId 用户ID
@@ -30,7 +32,8 @@ public class ChannelCache {
      */
     public void addCache(String userId, ChannelHandlerContext ctx) {
         USER_CHANNEL_MAP.put(userId, ctx);
-        log.info("用户 {} 的Channel已添加到缓存", userId);
+        reverseMap.put(ctx, userId);
+        log.info("用户 {} 的Channel已缓存", userId);
     }
 
     /**
@@ -44,11 +47,26 @@ public class ChannelCache {
 
     /**
      * 移除用户Channel映射
+     * @param ctx Channel上下文
+     */
+    public void removeCache(ChannelHandlerContext ctx) {
+        String userId = reverseMap.remove(ctx);
+        if (userId != null) {
+            USER_CHANNEL_MAP.remove(userId);
+            log.info("用户 {} 的Channel已移除", userId);
+        }
+    }
+
+    /**
+     * 移除用户Channel映射
      * @param userId 用户ID
      */
     public void removeCache(String userId) {
-        USER_CHANNEL_MAP.remove(userId);
-        log.info("用户 {} 的Channel已从缓存移除", userId);
+        ChannelHandlerContext ctx = USER_CHANNEL_MAP.remove(userId);
+        if (ctx != null) {
+            reverseMap.remove(ctx);
+            log.info("用户 {} 的Channel已移除", userId);
+        }
     }
 
     /**

src/main/java/com/zhentao/information/config/NettyConfig.java

@@ -76,10 +76,10 @@ public class NettyConfig {
                                 .addLast(new ChunkedWriteHandler())
                                 // HTTP消息聚合器
                                 .addLast(new HttpObjectAggregator(65536))
-                                // 心跳检测，60秒没有收到消息就触发
-                                .addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS))
+                                // 调整心跳检测时间：30秒没有收到消息就触发
+                                .addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS))
                                 // WebSocket协议处理器
-                                .addLast(new WebSocketServerProtocolHandler("/ws", null, true))
+                                .addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536))
                                 // 自定义消息处理器
                                 .addLast(webSocketHandler);
                     }

src/main/java/com/zhentao/information/controller/MessageController.java

@@ -1,6 +1,7 @@
 package com.zhentao.information.controller;
 
 import com.alibaba.fastjson.JSON;
+import com.zhentao.config.NullLogin;
 import com.zhentao.information.cache.ChannelCache;
 import com.zhentao.information.entity.ChatMessage;
 import com.zhentao.information.entity.Message;
@@ -32,6 +33,7 @@ public class MessageController {
      * 发送消息接口
      */
     @PostMapping("/send")
+    @NullLogin
     public String sendMessage(@RequestBody Message message) {
         log.info("收到消息：发送者={}, 接收者={}, 内容={}",
                 message.getFromUserId(),
@@ -70,6 +72,7 @@ public class MessageController {
      * 获取两个用户之间的聊天记录
      */
     @GetMapping("/history")
+    @NullLogin
     public List<ChatMessage> getChatHistory(@RequestHeader("token") String token, @RequestParam String userId2) {
         String userIdFromToken = TokenUtils.getUserIdFromToken(token);
         String chatId = generateChatId(userIdFromToken, userId2);
@@ -80,11 +83,22 @@ public class MessageController {
      * 获取用户的未读消息
      */
     @GetMapping("/unread")
+    @NullLogin
     public List<ChatMessage> getUnreadMessages(@RequestParam String userId) {
         return chatMessageRepository.findByToUserIdAndIsReadFalse(userId);
     }
 
     /**
+     * 查询某个群聊的所有消息
+     */
+    @GetMapping("/group/history")
+    @NullLogin
+    public List<ChatMessage> getGroupChatHistory(@RequestParam String groupId) {
+        // 查询chatId为group_群ID的所有消息
+        return chatMessageRepository.findByChatId("group_" + groupId);
+    }
+
+    /**
      * 生成聊天ID
      */
     private String generateChatId(String userId1, String userId2) {

src/main/java/com/zhentao/information/entity/Message.java

@@ -27,6 +27,11 @@ public class Message {
     private String toUserId;
     
     /**
+     * 群ID（群聊消息时使用）
+     */
+    private Long groupId;
+    
+    /**
      * 消息内容
      */
     private String content;

src/main/java/com/zhentao/information/handler/HeartbeatHandler.java

@@ -17,6 +17,9 @@ import org.springframework.stereotype.Component;
 @ChannelHandler.Sharable
 public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
 
+    private static final int MAX_MISSED_HEARTBEATS = 3;
+    private int missedHeartbeats = 0;
+
     /**
      * 处理用户事件
      * 当触发IdleStateEvent时调用
@@ -28,13 +31,46 @@ public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
         if (evt instanceof IdleStateEvent) {
             IdleStateEvent event = (IdleStateEvent) evt;
             
-            // 如果是读空闲事件
             if (event.state() == IdleState.READER_IDLE) {
-                log.info("读空闲，关闭连接：{}", ctx.channel().id().asLongText());
-                ctx.close();
+                missedHeartbeats++;
+                log.warn("读空闲，已错过 {} 次心跳", missedHeartbeats);
+                
+                if (missedHeartbeats >= MAX_MISSED_HEARTBEATS) {
+                    log.error("读空闲超时，关闭连接：{}", ctx.channel().id().asLongText());
+                    ctx.close();
+                } else {
+                    // 发送心跳检测消息
+                    ctx.writeAndFlush("ping");
+                }
+            } else if (event.state() == IdleState.WRITER_IDLE) {
+                // 发送心跳检测消息
+                ctx.writeAndFlush("ping");
             }
         } else {
             super.userEventTriggered(ctx, evt);
         }
     }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        if (msg instanceof String && "pong".equals(msg)) {
+            // 收到心跳响应，重置计数器
+            missedHeartbeats = 0;
+            log.debug("收到心跳响应");
+        } else {
+            super.channelRead(ctx, msg);
+        }
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        log.info("连接断开：{}", ctx.channel().id().asLongText());
+        super.channelInactive(ctx);
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        log.error("心跳处理器异常", cause);
+        ctx.close();
+    }
 } 

src/main/java/com/zhentao/information/handler/WebSocketHandler.java

@@ -1,10 +1,12 @@
 package com.zhentao.information.handler;
 
 import com.alibaba.fastjson.JSON;
+import com.zhentao.groups.MongoDB.pojo.GroupMessage;
+import com.zhentao.groups.MongoDB.pojo.Message;
 import com.zhentao.information.entity.ChatMessage;
-import com.zhentao.information.entity.Message;
 import com.zhentao.information.repository.ChatMessageRepository;
 import com.zhentao.information.service.WebSocketService;
+import com.zhentao.tool.TokenUtils;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
@@ -13,6 +15,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
+import java.util.Date;
 
 /**
  * WebSocket消息处理器
@@ -38,10 +41,35 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
         log.info("收到消息：{}", text);
         try {
             Message message = JSON.parseObject(text, Message.class);
-
+            log.info("接收到的消息：{}", message);
+            
             // 如果是连接消息，处理token
             if ("connect".equals(message.getType())) {
-                webSocketService.handleUserLogin(message.getContent(), ctx);
+                String userId = webSocketService.handleUserLogin(message.getContent(), ctx);
+                log.info("用户 {} 登录成功", userId);
+                if (userId != null) {
+                    // 用户登录成功后，自动加入所有群聊
+                    webSocketService.joinAllGroups(userId);
+                    // 发送连接成功消息
+                    Message response = new Message();
+                    response.setType("connect_success");
+                    response.setContent("连接成功");
+                    ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(response)));
+                }
+                return;
+            }
+
+            // 如果是心跳消息
+            if ("ping".equals(message.getType())) {
+                Message pongMessage = new Message();
+                pongMessage.setType("pong");
+                ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(pongMessage)));
+                return;
+            }
+
+            // 如果是群聊消息
+            if (message.getGroupId() != null) {
+                handleGroupMessage(message);
                 return;
             }
 
@@ -50,6 +78,28 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
 
         } catch (Exception e) {
             log.error("处理消息失败", e);
+            // 发送错误消息给客户端
+            Message errorMessage = new Message();
+            errorMessage.setType("error");
+            errorMessage.setContent("消息处理失败");
+            ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(errorMessage)));
+        }
+    }
+
+    /**
+     * 处理群聊消息
+     */
+    private void handleGroupMessage(Message message) {
+        // 设置群聊消息类型
+        message.setType("group_chat");
+        // 广播消息给群内所有成员
+        boolean sent = webSocketService.handleGroupMessage(message);
+        if (sent) {
+            // 发送消息确认
+            Message ackMessage = new Message();
+            ackMessage.setType("message_ack");
+            ackMessage.setContent("群消息已发送");
+            webSocketService.sendMessageToUser(message.getFromUserId(), ackMessage);
         }
     }
 
@@ -65,7 +115,7 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
         chatMessage.setFromUserId(message.getFromUserId());
         chatMessage.setToUserId(message.getToUserId());
         chatMessage.setContent(message.getContent());
-        chatMessage.setType(String.valueOf(message.getType()));
+        chatMessage.setType(message.getType());
         chatMessage.setTimestamp(System.currentTimeMillis());
         chatMessage.setIsRead(false);
         chatMessage.setChatId(chatId);
@@ -75,11 +125,20 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
 
         // 发送消息给接收者
         boolean sent = webSocketService.sendMessageToUser(message.getToUserId(), message);
-        System.err.println("判断对方用户是否在线"+sent);
         if (sent) {
             log.info("消息已发送给用户: {}, 内容: {}", message.getToUserId(), message.getContent());
+            // 发送消息确认给发送者
+            Message ackMessage = new Message();
+            ackMessage.setType("message_ack");
+            ackMessage.setContent("消息已发送");
+            webSocketService.sendMessageToUser(message.getFromUserId(), ackMessage);
         } else {
             log.info("用户 {} 不在线，消息已保存到MongoDB", message.getToUserId());
+            // 发送消息未送达通知给发送者
+            Message offlineMessage = new Message();
+            offlineMessage.setType("message_offline");
+            offlineMessage.setContent("对方不在线，消息已保存");
+            webSocketService.sendMessageToUser(message.getFromUserId(), offlineMessage);
         }
     }
 
@@ -97,6 +156,8 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
     @Override
     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
         log.info("连接断开：{}", ctx.channel().id().asLongText());
+        // 清理用户连接
+        webSocketService.removeUserConnection(ctx);
     }
 
     /**
@@ -112,8 +173,9 @@ public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketF
      * 生成聊天ID
      */
     private String generateChatId(String userId1, String userId2) {
+        // 确保两个用户之间的聊天ID唯一
         return userId1.compareTo(userId2) < 0 ?
-                userId1 + "_" + userId2 :
-                userId2 + "_" + userId1;
+            userId1 + "_" + userId2 :
+            userId2 + "_" + userId1;
     }
 }

src/main/java/com/zhentao/information/netty/NettyServer.java

@@ -17,6 +17,10 @@ import javax.annotation.Resource;
 
 /**
  * Netty服务器启动类
+ * 启动动一个基于 Netty 的 WebSocket 服务器
+ * 启动 WebSocket 服务器：
+ * 它会配置一个 Netty 的 ServerBootstrap，并绑定到指定的端口（默认是 8888）。
+ * 它会初始化一个 Netty 的通道管道（ChannelPipeline），并添加各种处理器来处理 WebSocket 连接和消息。
  */
 @Slf4j
 @Component

src/main/java/com/zhentao/information/repository/ChatMessageRepository.java

@@ -21,4 +21,9 @@ public interface ChatMessageRepository extends MongoRepository<ChatMessage, Stri
      * 查询用户的所有未读消息
      */
     List<ChatMessage> findByToUserIdAndIsReadFalse(String toUserId);
+
+    /**
+     * 查询toUserId为某个群ID的所有消息
+     */
+    List<ChatMessage> findByToUserId(String toUserId);
 }

src/main/java/com/zhentao/information/service/WebSocketService.java

@@ -1,17 +1,30 @@
 package com.zhentao.information.service;
 
 import com.alibaba.fastjson.JSON;
+import com.zhentao.groups.MongoDB.pojo.Message;
 import com.zhentao.information.cache.ChannelCache;
-import com.zhentao.information.entity.Message;
+import com.zhentao.information.cache.GroupChannelCache;
+import com.zhentao.information.cache.GroupMemberCache;
 import com.zhentao.tool.TokenUtils;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.group.ChannelGroup;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
+import com.zhentao.groups.service.GroupsService;
+import com.zhentao.groups.dto.GroupDto;
+import com.zhentao.information.entity.ChatMessage;
+import com.zhentao.information.repository.ChatMessageRepository;
 
 import javax.annotation.Resource;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /**
  * WebSocket服务类
@@ -24,8 +37,22 @@ public class WebSocketService {
     @Resource
     private ChannelCache channelCache;
 
+    @Resource
+    private GroupChannelCache groupChannelCache;
+
+    @Resource
+    private GroupMemberCache groupMemberCache;
+
+    @Autowired
+    @Lazy
+    private GroupsService groupsService;
+
+    @Autowired
+    private ChatMessageRepository chatMessageRepository;
+
     // 存储用户token的Map
     private final Map<String, String> userTokenMap = new ConcurrentHashMap<>();
+    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
 
     /**
      * 存储用户token
@@ -50,8 +77,9 @@ public class WebSocketService {
      * 处理用户登录
      * @param token 用户token
      * @param ctx Channel上下文
+     * @return 用户ID，如果登录失败返回null
      */
-    public void handleUserLogin(String token, ChannelHandlerContext ctx) {
+    public String handleUserLogin(String token, ChannelHandlerContext ctx) {
         String userId = TokenUtils.getUserIdFromToken(token);
         if (userId != null) {
             // 验证token是否与存储的token匹配
@@ -60,12 +88,13 @@ public class WebSocketService {
                 // 将用户ID和Channel绑定
                 channelCache.addCache(userId, ctx);
                 log.info("用户 {} 连接成功", userId);
-                
+
                 // 发送连接成功消息
                 Message response = new Message();
                 response.setType("connect_success");
                 response.setContent("连接成功");
                 ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(response)));
+                return userId;
             } else {
                 log.error("用户 {} 的token验证失败", userId);
                 ctx.close();
@@ -74,6 +103,7 @@ public class WebSocketService {
             log.error("无效的token");
             ctx.close();
         }
+        return null;
     }
 
     /**
@@ -83,12 +113,41 @@ public class WebSocketService {
      * @return 是否发送成功
      */
     public boolean sendMessageToUser(String userId, Message message) {
+        if (message.getType() == null) {
+            message.setType("text");
+        }
+        
         ChannelHandlerContext ctx = channelCache.getCache(userId);
-        if (ctx != null) {
-            ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(message)));
-            return true;
+        if (ctx != null && ctx.channel().isActive()) {
+            try {
+                String messageJson = JSON.toJSONString(message);
+                log.info("发送消息给用户 {}: {}", userId, messageJson);
+                ctx.channel().writeAndFlush(new TextWebSocketFrame(messageJson));
+                return true;
+            } catch (Exception e) {
+                log.error("发送消息给用户 {} 失败", userId, e);
+                retrySendMessage(userId, message);
+                return false;
+            }
+        } else {
+            log.info("用户 {} 不在线，消息将保存到数据库", userId);
+            return false;
         }
-        return false;
+    }
+
+    private void retrySendMessage(String userId, Message message) {
+        scheduler.schedule(() -> {
+            ChannelHandlerContext ctx = channelCache.getCache(userId);
+            if (ctx != null && ctx.channel().isActive()) {
+                try {
+                    String messageJson = JSON.toJSONString(message);
+                    log.info("重试发送消息给用户 {}: {}", userId, messageJson);
+                    ctx.channel().writeAndFlush(new TextWebSocketFrame(messageJson));
+                } catch (Exception e) {
+                    log.error("重试发送消息给用户 {} 失败", userId, e);
+                }
+            }
+        }, 1, TimeUnit.SECONDS);
     }
 
     /**
@@ -97,7 +156,15 @@ public class WebSocketService {
      */
     public void broadcastMessage(Message message) {
         channelCache.getAllCache().forEach((userId, ctx) -> {
-            ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(message)));
+            if (ctx.channel().isActive()) {
+                try {
+                    String messageJson = JSON.toJSONString(message);
+                    log.info("广播消息给用户 {}: {}", userId, messageJson);
+                    ctx.channel().writeAndFlush(new TextWebSocketFrame(messageJson));
+                } catch (Exception e) {
+                    log.error("广播消息给用户 {} 失败", userId, e);
+                }
+            }
         });
     }
 
@@ -107,6 +174,142 @@ public class WebSocketService {
      * @return 是否在线
      */
     public boolean isUserOnline(String userId) {
-        return channelCache.getCache(userId) != null;
+        ChannelHandlerContext ctx = channelCache.getCache(userId);
+        return ctx != null && ctx.channel().isActive();
+    }
+
+    /**
+     * 处理群聊消息
+     * @param message 群聊消息
+     * @return 是否发送成功
+     */
+    public boolean handleGroupMessage(Message message) {
+        Long groupId = message.getGroupId();
+        if (groupId == null) {
+            log.error("群聊消息缺少群ID");
+            return false;
+        }
+
+        // 存储群聊消息到MongoDB
+        ChatMessage chatMessage = new ChatMessage();
+        chatMessage.setFromUserId(message.getFromUserId());
+        chatMessage.setToUserId(String.valueOf(groupId));
+        chatMessage.setContent(message.getContent());
+        chatMessage.setType("group_chat");
+        chatMessage.setTimestamp(System.currentTimeMillis());
+        chatMessage.setIsRead(false);
+        chatMessage.setChatId("group_" + groupId);
+        chatMessageRepository.save(chatMessage);
+
+        // 获取群成员
+        List<GroupDto> groupList = groupsService.getList();
+        List<Long> groupMembers = null;
+        for (GroupDto group : groupList) {
+            if (group.getGroupId().equals(groupId)) {
+                groupMembers = group.getUid();
+                break;
+            }
+        }
+        
+        if (groupMembers == null || groupMembers.isEmpty()) {
+            log.error("群 {} 不存在或没有成员", groupId);
+            return false;
+        }
+
+        boolean allSent = true;
+        for (Long memberId : groupMembers) {
+            String memberIdStr = String.valueOf(memberId);
+            if (!memberIdStr.equals(message.getFromUserId())) {
+                ChannelHandlerContext ctx = channelCache.getCache(memberIdStr);
+                if (ctx != null && ctx.channel().isActive()) {
+                    try {
+                        String messageJson = JSON.toJSONString(message);
+                        log.info("发送群消息给用户 {}: {}", memberIdStr, messageJson);
+                        ctx.channel().writeAndFlush(new TextWebSocketFrame(messageJson));
+                    } catch (Exception e) {
+                        log.error("发送群消息给用户 {} 失败", memberId, e);
+                        allSent = false;
+                    }
+                }
+            }
+        }
+        log.info("群 {} 的消息已广播，群成员数：{}", groupId, groupMembers.size());
+        return allSent;
+    }
+
+    /**
+     * 用户登录时，将其加入所有群聊的ChannelGroup
+     * @param userId 用户ID
+     */
+    public void joinAllGroups(String userId) {
+        Long userIdLong = Long.valueOf(userId);
+        Map<Long, List<Long>> allGroups = groupMemberCache.getAllGroupMembers();
+
+        allGroups.forEach((groupId, members) -> {
+            if (members.contains(userIdLong)) {
+                addUserToGroup(groupId, userId);
+            }
+        });
+
+        log.info("用户 {} 已加入所有群聊的ChannelGroup", userId);
+    }
+
+    /**
+     * 将用户添加到群聊ChannelGroup
+     * @param groupId 群ID
+     * @param userId 用户ID
+     * @return 是否添加成功
+     */
+    public boolean addUserToGroup(Long groupId, String userId) {
+        // 验证用户是否在群中
+        if (!groupMemberCache.isUserInGroup(groupId, Long.valueOf(userId))) {
+            log.error("用户 {} 不在群 {} 中", userId, groupId);
+            return false;
+        }
+
+        ChannelHandlerContext ctx = channelCache.getCache(userId);
+        if (ctx == null || !ctx.channel().isActive()) {
+            log.error("用户 {} 不在线", userId);
+            return false;
+        }
+
+        ChannelGroup channelGroup = groupChannelCache.getGroup(groupId);
+        if (channelGroup == null) {
+            channelGroup = groupChannelCache.addGroup(groupId);
+        }
+
+        channelGroup.add(ctx.channel());
+        log.info("用户 {} 已添加到群 {} 的ChannelGroup", userId, groupId);
+        return true;
+    }
+
+    /**
+     * 将用户从群聊ChannelGroup中移除
+     * @param groupId 群ID
+     * @param userId 用户ID
+     * @return 是否移除成功
+     */
+    public boolean removeUserFromGroup(Long groupId, String userId) {
+        ChannelHandlerContext ctx = channelCache.getCache(userId);
+        if (ctx == null || !ctx.channel().isActive()) {
+            return false;
+        }
+
+        ChannelGroup channelGroup = groupChannelCache.getGroup(groupId);
+        if (channelGroup != null) {
+            channelGroup.remove(ctx.channel());
+            log.info("用户 {} 已从群 {} 的ChannelGroup移除", userId, groupId);
+            return true;
+        }
+        return false;
+    }
+
+    public void removeUserConnection(ChannelHandlerContext ctx) {
+        // 从ChannelCache中移除用户连接
+        channelCache.removeCache(ctx);
+        // 从所有群组中移除用户
+        groupChannelCache.getAllGroups().forEach((groupId, channelGroup) -> {
+            channelGroup.remove(ctx.channel());
+        });
     }
-} 
+}

src/main/java/com/zhentao/user/service/impl/UserLoginServiceImpl.java

@@ -21,6 +21,8 @@ import org.springframework.stereotype.Service;
 import org.springframework.util.DigestUtils;
 import org.springframework.web.multipart.MultipartFile;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 
@@ -221,8 +223,10 @@ public class UserLoginServiceImpl extends ServiceImpl<UserLoginMapper, UserLogin
                 // 将用户ID和token存储到WebSocketService中
                 webSocketService.storeUserToken(one.getId()+"", jwtToken);
 
-
-                return Result.OK("登录成功",jwtToken);
+                Map<String,Object> map = new HashMap<>();
+                map.put("token",jwtToken);
+                map.put("userId",one.getId()+"");
+                return Result.OK(map,"登录成功");
             }else {
                 // 如果获取锁超时，返回错误信息
                 return Result.ERR("获取锁超时",null);